From 1bd7726318005a9ed24a8c14b45311a05fa1a937 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 14 May 2024 22:50:15 +0200 Subject: [PATCH] WIP --- .../top10/{WordCount.java => Counter.java} | 4 +- .../wordcount/top10/Top10StreamProcessor.java | 6 +- .../de/juplo/kafka/wordcount/top10/Word.java | 2 + .../juplo/kafka/wordcount/top10/TestData.java | 150 +++++++++++------- 4 files changed, 98 insertions(+), 64 deletions(-) rename src/main/java/de/juplo/kafka/wordcount/top10/{WordCount.java => Counter.java} (91%) diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java b/src/main/java/de/juplo/kafka/wordcount/top10/Counter.java similarity index 91% rename from src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java rename to src/main/java/de/juplo/kafka/wordcount/top10/Counter.java index 8530f49..3dac384 100644 --- a/src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java +++ b/src/main/java/de/juplo/kafka/wordcount/top10/Counter.java @@ -14,9 +14,9 @@ import lombok.NoArgsConstructor; access = AccessLevel.PACKAGE) @Data @JsonIgnoreProperties(ignoreUnknown = true) -public class WordCount +public class Counter { String user; String word; - long count; + long counter; } diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java index 63357e8..b669344 100644 --- a/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java @@ -24,10 +24,10 @@ public class Top10StreamProcessor StreamsBuilder builder = new StreamsBuilder(); builder - .stream(inputTopic) - .map((word, wordCount) -> + .stream(inputTopic) + .map((word, counter) -> { - Entry entry = Entry.of(word.getWord(), wordCount.getCount()); + Entry entry = Entry.of(word.getWord(), counter.getCounter()); return new KeyValue<>(word.getUser(), entry); }) .groupByKey() diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Word.java b/src/main/java/de/juplo/kafka/wordcount/top10/Word.java index e4662bc..3ee1bb0 100644 --- a/src/main/java/de/juplo/kafka/wordcount/top10/Word.java +++ b/src/main/java/de/juplo/kafka/wordcount/top10/Word.java @@ -1,11 +1,13 @@ package de.juplo.kafka.wordcount.top10; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @NoArgsConstructor +@AllArgsConstructor(staticName = "of") @Data @JsonIgnoreProperties(ignoreUnknown = true) public class Word diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java index dd7ad33..f521869 100644 --- a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java @@ -1,51 +1,58 @@ package de.juplo.kafka.wordcount.top10; +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.streams.KeyValue; + import java.util.List; +import java.util.Map; +import java.util.Properties; import java.util.function.BiConsumer; +import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; class TestData { - static void writeInputData(BiConsumer consumer) + static void writeInputData(BiConsumer consumer) { consumer.accept( - "peter", - "{\"user\":\"peter\",\"word\":\"Hallo\"}"); + Word.of("peter","Hallo"), + Counter.of("peter","Hallo",1)); consumer.accept( - "klaus", - "{\"user\":\"klaus\",\"word\":\"Müsch\"}"); + Word.of("klaus","Müsch"), + Counter.of("klaus","Müsch",1)); consumer.accept( - "peter", - "{\"user\":\"peter\",\"word\":\"Welt\"}"); + Word.of("peter","Welt"), + Counter.of("peter","Welt",1)); consumer.accept( - "klaus", - "{\"user\":\"klaus\",\"word\":\"Müsch\"}"); + Word.of("klaus","Müsch"), + Counter.of("klaus","Müsch",2)); consumer.accept( - "klaus", - "{\"user\":\"klaus\",\"word\":\"s\"}"); + Word.of("klaus","s"), + Counter.of("klaus","s",1)); consumer.accept( - "peter", - "{\"user\":\"peter\",\"word\":\"Boäh\"}"); + Word.of("peter","Boäh"), + Counter.of("peter","Boäh",1)); consumer.accept( - "peter", - "{\"user\":\"peter\",\"word\":\"Welt\"}"); + Word.of("peter","Welt"), + Counter.of("peter","Welt",2)); consumer.accept( - "peter", - "{\"user\":\"peter\",\"word\":\"Boäh\"}"); + Word.of("peter","Boäh"), + Counter.of("peter","Boäh",2)); consumer.accept( - "klaus", - "{\"user\":\"klaus\",\"word\":\"s\"}"); + Word.of("klaus","s"), + Counter.of("klaus","s",2)); consumer.accept( - "peter", - "{\"user\":\"peter\",\"word\":\"Boäh\"}"); + Word.of("peter","Boäh"), + Counter.of("peter","Boäh",3)); consumer.accept( - "klaus", - "{\"user\":\"klaus\",\"word\":\"s\"}"); + Word.of("klaus","s"), + Counter.of("klaus","s",3)); } - static void assertExpectedResult(List receivedMessages) + static void assertExpectedResult(List> receivedMessages) { assertThat(receivedMessages).hasSize(11); assertThat(receivedMessages).containsSubsequence( @@ -66,40 +73,65 @@ class TestData expectedMessages[9]); // Boäh } - static Message[] expectedMessages = + static KeyValue[] expectedMessages = new KeyValue[] + { + KeyValue.pair( + Word.of("peter","Hallo"), + Counter.of("peter","Hallo",1)), + KeyValue.pair( + Word.of("klaus","Müsch"), + Counter.of("klaus","Müsch",1)), + KeyValue.pair( + Word.of("peter","Welt"), + Counter.of("peter","Welt",1)), + KeyValue.pair( + Word.of("klaus","Müsch"), + Counter.of("klaus","Müsch",2)), + KeyValue.pair( + Word.of("klaus","s"), + Counter.of("klaus","s",1)), + KeyValue.pair( + Word.of("peter","Boäh"), + Counter.of("peter","Boäh",1)), + KeyValue.pair( + Word.of("peter","Welt"), + Counter.of("peter","Welt",2)), + KeyValue.pair( + Word.of("peter","Boäh"), + Counter.of("peter","Boäh",2)), + KeyValue.pair( + Word.of("klaus","s"), + Counter.of("klaus","s",2)), + KeyValue.pair( + Word.of("peter","Boäh"), + Counter.of("peter","Boäh",3)), + KeyValue.pair( + Word.of("klaus","s"), + Counter.of("klaus","s",3)), + }; + + static Map convertToMap(Properties properties) + { + return properties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + } + + static String parseHeader(Headers headers, String key) { - Message.of( - "{\"user\":\"peter\",\"word\":\"Hallo\"}", - "{\"user\":\"peter\",\"word\":\"Hallo\",\"count\":1}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"Müsch\"}", - "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":1}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Welt\"}", - "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":1}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"Müsch\"}", - "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":2}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":1}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":1}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Welt\"}", - "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":2}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":2}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":2}"), - Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":3}"), - Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":3}"), - }; + Header header = headers.lastHeader(key); + if (header == null) + { + return key + "=null"; + } + else + { + return key + "=" + new String(header.value()); + } + } } -- 2.20.1