From c7f134403fb392077f24567d916a211949c7b197 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 16 Jun 2024 19:18:43 +0200 Subject: [PATCH] counter: 1.3.1 - Removed the defaults for serialization/deserialization --- .../counter/CounterApplicationConfiguriation.java | 2 -- .../wordcount/counter/CounterStreamProcessor.java | 11 +++++++---- 2 files changed, 7 insertions(+), 6 deletions(-) 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 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( -- 2.20.1