Spring Boot + Kafka, запись данных после обработки

@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));//.to("opapapa", Produced.with(Serdes.String().getClass(), UsersSerde.class));
 
        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;
                });//тут нужно отправить данные, которые обработал в kTable
 
        return kStream;
    }
}

kTable.toStream().to("tuktuktuk", Produced.with(Serdes.String(), usersAggregationSerde));