- 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.
</parent>
<groupId>de.juplo.kafka.wordcount</groupId>
<artifactId>counter</artifactId>
</parent>
<groupId>de.juplo.kafka.wordcount</groupId>
<artifactId>counter</artifactId>
- <version>1.1.11</version>
+ <version>1.2.0</version>
<name>Wordcount-Counter</name>
<description>Word-counting stream-processor of the multi-user wordcount-example</description>
<properties>
<name>Wordcount-Counter</name>
<description>Word-counting stream-processor of the multi-user wordcount-example</description>
<properties>
package de.juplo.kafka.wordcount.counter;
package de.juplo.kafka.wordcount.counter;
-import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
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;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
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.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;
import java.util.Properties;
{
StreamsBuilder builder = new StreamsBuilder();
{
StreamsBuilder builder = new StreamsBuilder();
- KStream<String, String> source = builder.stream(inputTopic);
+ KStream<String, Word> source = builder.stream(
+ inputTopic,
+ Consumed.with(
+ Serdes.String(),
+ new JsonSerde<>(Word.class)
+ .ignoreTypeHeaders()));
+
- .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
+ .<Word,Long>as(storeSupplier)
+ .withKeySerde(
+ new JsonSerde<>(Word.class)
+ .forKeys()
+ .noTypeInfo())
+ .withValueSerde(
+ Serdes.Long()))
+ .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();
+++ /dev/null
-package de.juplo.kafka.wordcount.counter;
-
-import lombok.Value;
-
-
-@Value(staticConstructor = "of")
-public class Key
-{
- private final String username;
- private final String word;
-}
--- /dev/null
+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;
+}
--- /dev/null
+package de.juplo.kafka.wordcount.counter;
+
+import lombok.Value;
+
+
+@Value(staticConstructor = "of")
+public class WordCount
+{
+ String user;
+ String word;
+ long count;
+}
{
consumer.accept(
"peter",
{
consumer.accept(
"peter",
+ "{\"user\":\"peter\",\"word\":\"Hallo\"}");
consumer.accept(
"klaus",
consumer.accept(
"klaus",
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}");
consumer.accept(
"peter",
consumer.accept(
"peter",
+ "{\"user\":\"peter\",\"word\":\"Welt\"}");
consumer.accept(
"klaus",
consumer.accept(
"klaus",
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}");
consumer.accept(
"klaus",
consumer.accept(
"klaus",
+ "{\"user\":\"klaus\",\"word\":\"s\"}");
consumer.accept(
"peter",
consumer.accept(
"peter",
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}");
consumer.accept(
"peter",
consumer.accept(
"peter",
+ "{\"user\":\"peter\",\"word\":\"Welt\"}");
consumer.accept(
"peter",
consumer.accept(
"peter",
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}");
consumer.accept(
"klaus",
consumer.accept(
"klaus",
+ "{\"user\":\"klaus\",\"word\":\"s\"}");
consumer.accept(
"peter",
consumer.accept(
"peter",
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}");
consumer.accept(
"klaus",
consumer.accept(
"klaus",
+ "{\"user\":\"klaus\",\"word\":\"s\"}");
}
static void assertExpectedResult(List<Message> receivedMessages)
}
static void assertExpectedResult(List<Message> receivedMessages)
static Message[] expectedMessages =
{
Message.of(
static Message[] expectedMessages =
{
Message.of(
- "{\"username\":\"peter\",\"word\":\"Hallo\"}",
- "1"),
+ "{\"user\":\"peter\",\"word\":\"Hallo\"}",
+ "{\"user\":\"peter\",\"word\":\"Hallo\",\"count\":1}"),
- "{\"username\":\"klaus\",\"word\":\"Müsch\"}",
- "1"),
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}",
+ "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":1}"),
- "{\"username\":\"peter\",\"word\":\"Welt\"}",
- "1"),
+ "{\"user\":\"peter\",\"word\":\"Welt\"}",
+ "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":1}"),
- "{\"username\":\"klaus\",\"word\":\"Müsch\"}",
- "2"),
+ "{\"user\":\"klaus\",\"word\":\"Müsch\"}",
+ "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":2}"),
- "{\"username\":\"klaus\",\"word\":\"s\"}",
- "1"),
+ "{\"user\":\"klaus\",\"word\":\"s\"}",
+ "{\"user\":\"klaus\",\"word\":\"s\",\"count\":1}"),
- "{\"username\":\"peter\",\"word\":\"Boäh\"}",
- "1"),
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}",
+ "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":1}"),
- "{\"username\":\"peter\",\"word\":\"Welt\"}",
- "2"),
+ "{\"user\":\"peter\",\"word\":\"Welt\"}",
+ "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":2}"),
- "{\"username\":\"peter\",\"word\":\"Boäh\"}",
- "2"),
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}",
+ "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":2}"),
- "{\"username\":\"klaus\",\"word\":\"s\"}",
- "2"),
+ "{\"user\":\"klaus\",\"word\":\"s\"}",
+ "{\"user\":\"klaus\",\"word\":\"s\",\"count\":2}"),
- "{\"username\":\"peter\",\"word\":\"Boäh\"}",
- "3"),
+ "{\"user\":\"peter\",\"word\":\"Boäh\"}",
+ "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":3}"),
- "{\"username\":\"klaus\",\"word\":\"s\"}",
- "3"),
+ "{\"user\":\"klaus\",\"word\":\"s\"}",
+ "{\"user\":\"klaus\",\"word\":\"s\",\"count\":3}"),