From: Kai Moritz <kai@juplo.de>
Date: Tue, 14 May 2024 22:12:47 +0000 (+0200)
Subject: WIP
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;ds=sidebyside;h=58767ef9d9dd8841d28883567ac169b46e450761;p=demos%2Fkafka%2Fwordcount

WIP
---

diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java
index 7a03ba0..58fb65f 100644
--- a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java
+++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java
@@ -41,17 +41,19 @@ public class Top10StreamProcessorTopologyTest
 
     TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
 
-    TestInputTopic<Key, Counter> in = testDriver.createInputTopic(
+    TestInputTopic<Key, Entry> in = testDriver.createInputTopic(
         IN,
         (JsonSerializer<Key>)keySerde.serializer(),
-        (JsonSerializer<Counter>)valueSerde.serializer());
+        (JsonSerializer<Entry>)valueSerde.serializer());
 
     TestOutputTopic<String, Ranking> out = testDriver.createOutputTopic(
         OUT,
         (JsonDeserializer<String>)keySerde.deserializer(),
         (JsonDeserializer<Ranking>)valueSerde.deserializer());
 
-    TestData.writeInputData((key, value) -> in.pipeInput(key, value));
+    TestData.writeInputData((key, value) -> in.pipeInput(
+        key,
+        Entry.of(value.getWord(), value.getCounter())));
 
     List<KeyValue<String, Ranking>> receivedMessages = out
         .readRecordsToList()