1 package de.juplo.kafka.wordcount.counter;
3 import de.juplo.kafka.wordcount.splitter.TestInputUser;
4 import de.juplo.kafka.wordcount.splitter.TestInputWord;
5 import de.juplo.kafka.wordcount.top10.TestOutputWord;
6 import de.juplo.kafka.wordcount.top10.TestOutputWordCounter;
7 import lombok.extern.slf4j.Slf4j;
8 import org.apache.kafka.streams.TestInputTopic;
9 import org.apache.kafka.streams.TestOutputTopic;
10 import org.apache.kafka.streams.Topology;
11 import org.apache.kafka.streams.TopologyTestDriver;
12 import org.apache.kafka.streams.state.KeyValueStore;
13 import org.apache.kafka.streams.state.Stores;
14 import org.junit.jupiter.api.AfterEach;
15 import org.junit.jupiter.api.BeforeEach;
16 import org.junit.jupiter.api.Test;
17 import org.springframework.kafka.support.serializer.JsonDeserializer;
18 import org.springframework.kafka.support.serializer.JsonSerializer;
19 import org.springframework.util.LinkedMultiValueMap;
20 import org.springframework.util.MultiValueMap;
24 import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
25 import static de.juplo.kafka.wordcount.counter.CounterStreamProcessor.STORE_NAME;
29 public class CounterStreamProcessorTopologyTest
31 public static final String IN = "TEST-IN";
32 public static final String OUT = "TEST-OUT";
35 TopologyTestDriver testDriver;
36 TestInputTopic<TestInputUser, TestInputWord> in;
37 TestOutputTopic<TestOutputWord, TestOutputWordCounter> out;
41 public void setUpTestDriver()
43 Topology topology = CounterStreamProcessor.buildTopology(
46 Stores.inMemoryKeyValueStore(STORE_NAME));
48 testDriver = new TopologyTestDriver(topology, serializationConfig());
50 in = testDriver.createInputTopic(IN, serializer(), serializer());
51 out = testDriver.createOutputTopic(OUT, keyDeserializer(), valueDeserializer());
60 .forEach(kv -> in.pipeInput(kv.key, kv.value));
62 MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
65 .forEach(record -> receivedMessages.add(record.key(), record.value()));
67 TestData.assertExpectedMessages(receivedMessages);
69 TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
70 TestData.assertExpectedLastMessagesForWord(receivedMessages);
72 KeyValueStore<Word, Long> store = testDriver.getKeyValueStore(STORE_NAME);
73 TestData.assertExpectedState(store);
77 public void tearDown()
83 private static JsonSerializer serializer()
85 return new JsonSerializer().noTypeInfo();
88 private JsonDeserializer<TestOutputWord> keyDeserializer()
90 return deserializer(true);
93 private static JsonDeserializer<TestOutputWordCounter> valueDeserializer()
95 return deserializer(false);
98 private static <T> JsonDeserializer<T> deserializer(boolean isKey)
100 JsonDeserializer<T> deserializer = new JsonDeserializer<>();
101 deserializer.configure(
102 Map.of(JsonDeserializer.TYPE_MAPPINGS, typeMappingsConfig()),
107 private static String typeMappingsConfig()
109 return CounterStreamProcessor.typeMappingsConfig(TestOutputWord.class, TestOutputWordCounter.class);