From: Kai Moritz <kai@juplo.de>
Date: Sun, 16 Jun 2024 17:18:43 +0000 (+0200)
Subject: counter: 1.3.1 - Removed the defaults for serialization/deserialization
X-Git-Tag: counter-1.3.1~4
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=c7f134403fb392077f24567d916a211949c7b197;p=demos%2Fkafka%2Fwordcount

counter: 1.3.1 - Removed the defaults for serialization/deserialization
---

diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java
index 484b8de..d9579a5 100644
--- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java
+++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java
@@ -50,8 +50,6 @@ public class CounterApplicationConfiguriation
 
 		propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
 		propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName());
-		propertyMap.put(JsonDeserializer.KEY_DEFAULT_TYPE, User.class.getName());
-		propertyMap.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Word.class.getName());
 		propertyMap.put(
 				JsonDeserializer.TYPE_MAPPINGS,
 				"user:" + User.class.getName() + "," +
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(