From b9f05dfc77389331dbc69bf34898231074eb13d5 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Tue, 14 May 2024 22:45:17 +0200
Subject: [PATCH] WIP

---
 pom.xml                                       |  21 +++-
 .../de/juplo/kafka/wordcount/top10/Entry.java |  10 +-
 .../kafka/wordcount/top10/WordCount.java      |   5 +
 .../juplo/kafka/wordcount/top10/TestData.java | 105 ++++++++++++++++++
 4 files changed, 135 insertions(+), 6 deletions(-)
 create mode 100644 src/test/java/de/juplo/kafka/wordcount/top10/TestData.java

diff --git a/pom.xml b/pom.xml
index 6ccc687..746bf28 100644
--- a/pom.xml
+++ b/pom.xml
@@ -15,8 +15,6 @@
 	<description>Top-10 stream-processor of the multi-user wordcount-example</description>
 	<properties>
 		<docker-maven-plugin.version>0.33.0</docker-maven-plugin.version>
-		<java.version>11</java.version>
-		<kafka.version>2.8.0</kafka.version>
 	</properties>
 	<dependencies>
 		<dependency>
@@ -52,15 +50,34 @@
 			<artifactId>lombok</artifactId>
 			<optional>true</optional>
 		</dependency>
+
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.kafka</groupId>
+			<artifactId>spring-kafka-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.assertj</groupId>
+			<artifactId>assertj-core</artifactId>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>
 		<plugins>
+			<plugin>
+				<artifactId>maven-failsafe-plugin</artifactId>
+			</plugin>
 			<plugin>
 				<groupId>org.springframework.boot</groupId>
 				<artifactId>spring-boot-maven-plugin</artifactId>
diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Entry.java b/src/main/java/de/juplo/kafka/wordcount/top10/Entry.java
index 67f45f2..9f4e8ce 100644
--- a/src/main/java/de/juplo/kafka/wordcount/top10/Entry.java
+++ b/src/main/java/de/juplo/kafka/wordcount/top10/Entry.java
@@ -1,11 +1,13 @@
 package de.juplo.kafka.wordcount.top10;
 
-import lombok.Value;
+import lombok.AllArgsConstructor;
+import lombok.Data;
 
 
-@Value(staticConstructor = "of")
+@Data
+@AllArgsConstructor(staticName = "of")
 public class Entry
 {
-  private final String word;
-  private final Long count;
+  private String word;
+  private Long count;
 }
diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java b/src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java
index 69c8bed..8530f49 100644
--- a/src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java
+++ b/src/main/java/de/juplo/kafka/wordcount/top10/WordCount.java
@@ -2,11 +2,16 @@ package de.juplo.kafka.wordcount.top10;
 
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
 
 @NoArgsConstructor
+@AllArgsConstructor(
+    staticName = "of",
+    access = AccessLevel.PACKAGE)
 @Data
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class WordCount
diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java
new file mode 100644
index 0000000..dd7ad33
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/top10/TestData.java
@@ -0,0 +1,105 @@
+package de.juplo.kafka.wordcount.top10;
+
+import java.util.List;
+import java.util.function.BiConsumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+class TestData
+{
+	static void writeInputData(BiConsumer<String, String> consumer)
+	{
+		consumer.accept(
+				"peter",
+				"{\"user\":\"peter\",\"word\":\"Hallo\"}");
+		consumer.accept(
+				"klaus",
+				"{\"user\":\"klaus\",\"word\":\"Müsch\"}");
+		consumer.accept(
+				"peter",
+				"{\"user\":\"peter\",\"word\":\"Welt\"}");
+		consumer.accept(
+				"klaus",
+				"{\"user\":\"klaus\",\"word\":\"Müsch\"}");
+		consumer.accept(
+				"klaus",
+				"{\"user\":\"klaus\",\"word\":\"s\"}");
+		consumer.accept(
+				"peter",
+				"{\"user\":\"peter\",\"word\":\"Boäh\"}");
+		consumer.accept(
+				"peter",
+				"{\"user\":\"peter\",\"word\":\"Welt\"}");
+		consumer.accept(
+				"peter",
+				"{\"user\":\"peter\",\"word\":\"Boäh\"}");
+		consumer.accept(
+				"klaus",
+				"{\"user\":\"klaus\",\"word\":\"s\"}");
+		consumer.accept(
+				"peter",
+				"{\"user\":\"peter\",\"word\":\"Boäh\"}");
+		consumer.accept(
+				"klaus",
+				"{\"user\":\"klaus\",\"word\":\"s\"}");
+	}
+
+	static void assertExpectedResult(List<Message> receivedMessages)
+	{
+		assertThat(receivedMessages).hasSize(11);
+		assertThat(receivedMessages).containsSubsequence(
+				expectedMessages[0]); // Hallo
+		assertThat(receivedMessages).containsSubsequence(
+				expectedMessages[1],
+				expectedMessages[3]); // Müsch
+		assertThat(receivedMessages).containsSubsequence(
+				expectedMessages[2],
+				expectedMessages[6]);
+		assertThat(receivedMessages).containsSubsequence(
+				expectedMessages[4],
+				expectedMessages[8],
+				expectedMessages[10]); // s
+		assertThat(receivedMessages).containsSubsequence(
+				expectedMessages[5],
+				expectedMessages[7],
+				expectedMessages[9]); // Boäh
+	}
+
+	static Message[] expectedMessages =
+	{
+			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}"),
+	};
+}
-- 
2.20.1