From: Kai Moritz <kai@juplo.de>
Date: Mon, 17 Jun 2024 18:43:04 +0000 (+0200)
Subject: popular: 1.0.0 - Added ``peek()``-operations
X-Git-Tag: popularl-1.0.0
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=e88db15eafe64ac083d066be29ad94058930bb5d;p=demos%2Fkafka%2Fwordcount

popular: 1.0.0 - Added ``peek()``-operations
---

diff --git a/src/main/java/de/juplo/kafka/wordcount/popular/PopularStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/popular/PopularStreamProcessor.java
index 9ba9ad2..788ae20 100644
--- a/src/main/java/de/juplo/kafka/wordcount/popular/PopularStreamProcessor.java
+++ b/src/main/java/de/juplo/kafka/wordcount/popular/PopularStreamProcessor.java
@@ -62,7 +62,9 @@ public class PopularStreamProcessor
 
 		builder
 				.stream(inputTopic, Consumed.with(inKeySerde(), inValueSerde()))
+				.peek((user, userWord) -> log.info("{}: {} -> {}", inputTopic, user, userWord))
 				.map((key, userWord) -> new KeyValue<>(Word.of(userWord.getWord()), Word.of(userWord.getWord())))
+				.peek((key, value) -> log.info("mapped: {} -> {}", key, value))
 				.groupByKey()
 				.windowedBy(TimeWindows.ofSizeWithNoGrace(WINDOW_SIZE))
 				.count(
@@ -71,18 +73,21 @@ public class PopularStreamProcessor
 								.withKeySerde(new JsonSerde<>(Word.class).noTypeInfo())
 								.withValueSerde(Serdes.Long()))
 				.toStream()
+				.peek((windowedWord, counter) -> log.info("windowed: {} -> {}", windowedWord, counter))
 				.map((windowedWord, counter) -> new KeyValue<>(
 						WindowedWord.of(
 								ZonedDateTime.ofInstant(windowedWord.window().startTime(), zone),
 								ZonedDateTime.ofInstant(windowedWord.window().endTime(), zone),
 								windowedWord.key().getWord()),
 						WordCounter.of(windowedWord.key().getWord(), counter)))
+				.peek((windowedWord, wordCounter) -> log.info("results: {} -> {}", windowedWord, wordCounter))
 				.toTable(
 						Materialized
 								.<WindowedWord, WordCounter>as(keyValueBytesStoreSupplier)
 								.withKeySerde(new JsonSerde<>(WindowedWord.class).noTypeInfo())
 								.withValueSerde(new JsonSerde<>(WordCounter.class).noTypeInfo()))
 				.toStream()
+				.peek((windowedWord, wordCounter) -> log.info("output: {} -> {}", windowedWord, wordCounter))
 				.to(outputTopic, Produced.with(outKeySerde(), outValueSerde()));
 
 		Topology topology = builder.build();