X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterStreamProcessorTopologyTest.java;h=6e244e26c4d3c6ed8043f820a968c7a8a17b40ca;hb=8d0426539d69616900f4d6ef19e52d50b497f57f;hp=e5964dc7eebc35d21d2e03c95804734775187889;hpb=3c23bfd42005211ac9812fba698ab74c8a6b7aa0;p=demos%2Fkafka%2Fwordcount

diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
index e5964dc..6e244e2 100644
--- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
+++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java
@@ -9,6 +9,7 @@ import org.apache.kafka.streams.TestInputTopic;
 import org.apache.kafka.streams.TestOutputTopic;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.Stores;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -19,16 +20,14 @@ import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 
 import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
-import static de.juplo.kafka.wordcount.counter.TestData.parseHeader;
-import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
-import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
 
 
 @Slf4j
 public class CounterStreamProcessorTopologyTest
 {
-  public final static String IN = "TEST-IN";
-  public final static String OUT = "TEST-OUT";
+  public static final String IN = "TEST-IN";
+  public static final String OUT = "TEST-OUT";
+  public static final String STORE_NAME = "TOPOLOGY-TEST";
 
 
   TopologyTestDriver testDriver;
@@ -42,7 +41,7 @@ public class CounterStreamProcessorTopologyTest
     Topology topology = CounterStreamProcessor.buildTopology(
         IN,
         OUT,
-        Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
+        Stores.inMemoryKeyValueStore(STORE_NAME));
 
     testDriver = new TopologyTestDriver(topology, serializationConfig());
 
@@ -67,23 +66,20 @@ public class CounterStreamProcessorTopologyTest
   {
     TestData
         .getInputMessages()
-        .forEach(word -> in.pipeInput(word.getUser(), word));
+        .forEach(kv -> in.pipeInput(kv.key, kv.value));
 
     MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
     out
         .readRecordsToList()
-        .forEach(record ->
-        {
-          log.debug(
-              "OUT: {} -> {}, {}, {}",
-              record.key(),
-              record.value(),
-              parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
-              parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
-          receivedMessages.add(record.key(), record.value());
-        });
+        .forEach(record -> receivedMessages.add(record.key(), record.value()));
 
     TestData.assertExpectedMessages(receivedMessages);
+
+    TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
+    TestData.assertExpectedLastMessagesForWord(receivedMessages);
+
+    KeyValueStore<Word, Long> store = testDriver.getKeyValueStore(STORE_NAME);
+    TestData.assertExpectedState(store);
   }
 
   @AfterEach