How can I send a message to a topic after processing in kTable, kafka, java

Я прошу прощения за объяснение.

Я сделал группировку и вызвал функцию, после этого мне нужно отправить весь результат в тему, но если я попытаюсь прописать toStream и (...) он попросит меня сделать Serde, он тоже это сделал, но он не нравится.

Может я где-то не писал или есть другой способ отправить сообщение в тему после обработки данных?

(мне нужно отправить обработанные данные), спасибо.

@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaConsumerConfig {

    private static final String TOPIC = "All_Users1234449313123q4r345";

    private static final String strAdmin = "topicAdmin";

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs(KafkaProperties kafkaProperties) {
        Map<String, Object> config = new HashMap<>();
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "tak-takisqrt");
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, UsersSerde.class);
        config.put("default.deserialization.exception.handler", LogAndContinueExceptionHandler.class);
        //config.put("default.deserialization.exception.handler", LogAndFailExceptionHandler.class);
        return new KafkaStreamsConfiguration(config);
    }

    @Bean
    public KStream<String, Users> kStream(StreamsBuilder kStreamBuilder) {
        Serde<String> stringSerde = new Serdes.StringSerde();
        Serde<Users> usersSerde = new UsersSerde();
        Serde<UsersAggregation> usersAggregationSerde = new UsersAggregationSerde();

        KStream<String, Users> kStream = kStreamBuilder
                .stream(TOPIC, Consumed.with(stringSerde,
                        usersSerde));

        KTable<String, UsersAggregation>  kTable = kStream
                .filter((k,v) -> v.getRole().compareTo("Admin") == 0)
                .selectKey((a, b) -> b.getDate() + b.getName() + b.getRole())
                .groupByKey()
                .aggregate(UsersAggregation::new,
                        (k, v, agg) -> {
                            agg.setName(v.getName());
                            agg.setDate(v.getDate());
                            agg.setRole(v.getRole());
                            agg.setCount(agg.getCount() + 1);
                            agg.setActions(v.getAction());
                            return agg;
                            //});
                        }, Materialized.with(stringSerde, usersAggregationSerde))
                .mapValues(v -> {
                    v.sortActions();
                    System.out.println(v.toString());
                    return v;
                });

        return kStream;
    }
}

После этих шагов я должен опубликовать все это в новой теме, но я не знаю, как разместить.