+ .map((word, count) -> new KeyValue<>(word, WordCount.of(word.getUser(), word.getWord(), count)))
+ .to(
+ outputTopic,
+ Produced.with(
+ new JsonSerde<>(Word.class)
+ .forKeys()
+ .noTypeInfo(),
+ new JsonSerde<>(WordCount.class)
+ .noTypeInfo()));
+
+ Topology topology = builder.build();
+ log.info("\n\n{}", topology.describe());