WIP

diff --git a/pom.xml b/pom.xml
index 6ccc687..746bf28 100644 (file)
--- 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 (file)
--- 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 (file)
--- 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 (file)
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}"),
+       };
+}