From: Kai Moritz Date: Wed, 15 Feb 2023 07:19:09 +0000 (+0100) Subject: counter: 1.2.0 - Service reads/writes using `JsonSerde` X-Git-Tag: counter-1.2.0 X-Git-Url: https://juplo.de/gitweb/?p=demos%2Fkafka%2Fwordcount;a=commitdiff_plain;h=c2d06f0efd9e09a9df1c4c536d54ad969cf456eb counter: 1.2.0 - Service reads/writes using `JsonSerde` - First working version of a refined topology, that reads and writes key and value with the `JsonSerde`. - This topology expects and writes all data as JSON, except the key of the incomming messages. - All used Serdes are directly configured in the topology. --- diff --git a/pom.xml b/pom.xml index 3b72889..ccdd0fc 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount counter - 1.1.11 + 1.2.0 Wordcount-Counter Word-counting stream-processor of the multi-user wordcount-example 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 4002df2..324e424 100644 --- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java @@ -1,8 +1,8 @@ package de.juplo.kafka.wordcount.counter; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; @@ -10,6 +10,8 @@ import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; +import org.apache.kafka.streams.kstream.*; +import org.springframework.kafka.support.serializer.JsonSerde; import java.util.Properties; @@ -44,27 +46,44 @@ public class CounterStreamProcessor { StreamsBuilder builder = new StreamsBuilder(); - KStream source = builder.stream(inputTopic); + KStream source = builder.stream( + inputTopic, + Consumed.with( + Serdes.String(), + new JsonSerde<>(Word.class) + .ignoreTypeHeaders())); + source - .map((username, word) -> - { - try - { - String key = mapper.writeValueAsString(Key.of(username, word)); - return new KeyValue<>(key, word); - } - catch (JsonProcessingException e) - { - throw new RuntimeException(e); - } - }) - .groupByKey() - .count(Materialized.as(storeSupplier)) - .mapValues(value->Long.toString(value)) + .map((key, word) -> new KeyValue<>(word, word)) + .groupByKey(Grouped.with( + new JsonSerde<>(Word.class) + .forKeys() + .noTypeInfo(), + new JsonSerde<>(Word.class) + .noTypeInfo())) + .count(Materialized + .as(storeSupplier) + .withKeySerde( + new JsonSerde<>(Word.class) + .forKeys() + .noTypeInfo()) + .withValueSerde( + Serdes.Long())) .toStream() - .to(outputTopic); + .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()); - return builder.build(); + return topology; } public void start() diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/Key.java b/src/main/java/de/juplo/kafka/wordcount/counter/Key.java deleted file mode 100644 index 1e00dca..0000000 --- a/src/main/java/de/juplo/kafka/wordcount/counter/Key.java +++ /dev/null @@ -1,11 +0,0 @@ -package de.juplo.kafka.wordcount.counter; - -import lombok.Value; - - -@Value(staticConstructor = "of") -public class Key -{ - private final String username; - private final String word; -} diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/Word.java b/src/main/java/de/juplo/kafka/wordcount/counter/Word.java new file mode 100644 index 0000000..4aa5ee2 --- /dev/null +++ b/src/main/java/de/juplo/kafka/wordcount/counter/Word.java @@ -0,0 +1,17 @@ +package de.juplo.kafka.wordcount.counter; + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + + +@AllArgsConstructor(staticName = "of") +@NoArgsConstructor +@Data +@JsonIgnoreProperties(ignoreUnknown = true) +public class Word +{ + private String user; + private String word; +} diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java b/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java new file mode 100644 index 0000000..44ccb2d --- /dev/null +++ b/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java @@ -0,0 +1,12 @@ +package de.juplo.kafka.wordcount.counter; + +import lombok.Value; + + +@Value(staticConstructor = "of") +public class WordCount +{ + String user; + String word; + long count; +} diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java index dc3bcec..8ff7022 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -12,37 +12,37 @@ class TestData { consumer.accept( "peter", - "Hallo"); + "{\"user\":\"peter\",\"word\":\"Hallo\"}"); consumer.accept( "klaus", - "Müsch"); + "{\"user\":\"klaus\",\"word\":\"Müsch\"}"); consumer.accept( "peter", - "Welt"); + "{\"user\":\"peter\",\"word\":\"Welt\"}"); consumer.accept( "klaus", - "Müsch"); + "{\"user\":\"klaus\",\"word\":\"Müsch\"}"); consumer.accept( "klaus", - "s"); + "{\"user\":\"klaus\",\"word\":\"s\"}"); consumer.accept( "peter", - "Boäh"); + "{\"user\":\"peter\",\"word\":\"Boäh\"}"); consumer.accept( "peter", - "Welt"); + "{\"user\":\"peter\",\"word\":\"Welt\"}"); consumer.accept( "peter", - "Boäh"); + "{\"user\":\"peter\",\"word\":\"Boäh\"}"); consumer.accept( "klaus", - "s"); + "{\"user\":\"klaus\",\"word\":\"s\"}"); consumer.accept( "peter", - "Boäh"); + "{\"user\":\"peter\",\"word\":\"Boäh\"}"); consumer.accept( "klaus", - "s"); + "{\"user\":\"klaus\",\"word\":\"s\"}"); } static void assertExpectedResult(List receivedMessages) @@ -69,37 +69,37 @@ class TestData static Message[] expectedMessages = { Message.of( - "{\"username\":\"peter\",\"word\":\"Hallo\"}", - "1"), + "{\"user\":\"peter\",\"word\":\"Hallo\"}", + "{\"user\":\"peter\",\"word\":\"Hallo\",\"count\":1}"), Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "1"), + "{\"user\":\"klaus\",\"word\":\"Müsch\"}", + "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":1}"), Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "1"), + "{\"user\":\"peter\",\"word\":\"Welt\"}", + "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":1}"), Message.of( - "{\"username\":\"klaus\",\"word\":\"Müsch\"}", - "2"), + "{\"user\":\"klaus\",\"word\":\"Müsch\"}", + "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":2}"), Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "1"), + "{\"user\":\"klaus\",\"word\":\"s\"}", + "{\"user\":\"klaus\",\"word\":\"s\",\"count\":1}"), Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "1"), + "{\"user\":\"peter\",\"word\":\"Boäh\"}", + "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":1}"), Message.of( - "{\"username\":\"peter\",\"word\":\"Welt\"}", - "2"), + "{\"user\":\"peter\",\"word\":\"Welt\"}", + "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":2}"), Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "2"), + "{\"user\":\"peter\",\"word\":\"Boäh\"}", + "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":2}"), Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "2"), + "{\"user\":\"klaus\",\"word\":\"s\"}", + "{\"user\":\"klaus\",\"word\":\"s\",\"count\":2}"), Message.of( - "{\"username\":\"peter\",\"word\":\"Boäh\"}", - "3"), + "{\"user\":\"peter\",\"word\":\"Boäh\"}", + "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":3}"), Message.of( - "{\"username\":\"klaus\",\"word\":\"s\"}", - "3"), + "{\"user\":\"klaus\",\"word\":\"s\"}", + "{\"user\":\"klaus\",\"word\":\"s\",\"count\":3}"), }; }