X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterStreamProcessor.java;h=8b9c12b2611f6a4dddc9c813e1c3ace403a6a803;hb=c7f134403fb392077f24567d916a211949c7b197;hp=64bd619d0d30d5e021e5d3be3cc59769737a3831;hpb=d8173052504e89f85b09ce060302e87979973714;p=demos%2Fkafka%2Fwordcount

diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
index 64bd619..8b9c12b 100644
--- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
+++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java
@@ -2,7 +2,7 @@ package de.juplo.kafka.wordcount.counter;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.streams.*;
-import org.apache.kafka.streams.kstream.KStream;
+import org.apache.kafka.streams.kstream.Consumed;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
 import org.apache.kafka.streams.state.QueryableStoreTypes;
@@ -42,9 +42,12 @@ public class CounterStreamProcessor
 	{
 		StreamsBuilder builder = new StreamsBuilder();
 
-		KStream<User, Word> source = builder.stream(inputTopic);
-
-		source
+		builder
+				.stream(
+						inputTopic,
+						Consumed.with(
+								new JsonSerde<>(User.class),
+								new JsonSerde<>(Word.class)))
 				.map((key, word) -> new KeyValue<>(word, word))
 				.groupByKey()
 				.count(