From: Kai Moritz <kai@juplo.de>
Date: Tue, 14 May 2024 21:48:34 +0000 (+0200)
Subject: WIP
X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=e7c52491bcef70efd85487b7469a32e832d61724;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
new file mode 100644
index 0000000..8b4a593
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java
@@ -0,0 +1,77 @@
+package de.juplo.kafka.wordcount.top10;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.streams.*;
+import org.apache.kafka.streams.state.Stores;
+import org.junit.jupiter.api.Test;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerde;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static de.juplo.kafka.wordcount.counter.TestData.convertToMap;
+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 Top10StreamProcessorTopologyTest
+{
+  public final static String IN = "TEST-IN";
+  public final static String OUT = "TEST-OUT";
+
+  @Test
+  public void test()
+  {
+    Topology topology = CounterStreamProcessor.buildTopology(
+        IN,
+        OUT,
+        Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
+
+    CounterApplicationConfiguriation applicationConfiguriation =
+        new CounterApplicationConfiguriation();
+    Properties streamProcessorProperties =
+        applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
+    Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
+
+    JsonSerde<?> keySerde = new JsonSerde<>();
+    keySerde.configure(propertyMap, true);
+    JsonSerde<?> valueSerde = new JsonSerde<>();
+    valueSerde.configure(propertyMap, false);
+
+    TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
+
+    TestInputTopic<String, Word> in = testDriver.createInputTopic(
+        IN,
+        (JsonSerializer<String>)keySerde.serializer(),
+        (JsonSerializer<Word>)valueSerde.serializer());
+
+    TestOutputTopic<Word, WordCounter> out = testDriver.createOutputTopic(
+        OUT,
+        (JsonDeserializer<Word>)keySerde.deserializer(),
+        (JsonDeserializer<WordCounter>)valueSerde.deserializer());
+
+    TestData.writeInputData((key, value) -> in.pipeInput(key, value));
+
+    List<KeyValue<Word, WordCounter>> receivedMessages = out
+        .readRecordsToList()
+        .stream()
+        .map(record ->
+        {
+          log.debug(
+              "OUT: {} -> {}, {}, {}",
+              record.key(),
+              record.value(),
+              parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
+              parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
+          return KeyValue.pair(record.key(), record.value());
+        })
+        .toList();
+
+    TestData.assertExpectedResult(receivedMessages);
+  }
+}