X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Ftop10%2FTop10StreamProcessorTopologyTest.java;h=58fb65fda21b9630c7d73e475f8bce3fb6c5372c;hb=20da5a144f8478fd7a00081c39f34a6670a68101;hp=7a03ba01d6be8b8470d558c30a630243b31ce72b;hpb=32bd4b205c027ce14fd694b036a811d2bfed7042;p=demos%2Fkafka%2Fwordcount

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()