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;
23 import java.util.stream.Collectors;
25 import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
26 import static de.juplo.kafka.wordcount.counter.CounterStreamProcessor.STORE_NAME;
30 public class CounterStreamProcessorTopologyTest
32 public static final String IN = "TEST-IN";
33 public static final String OUT = "TEST-OUT";
36 TopologyTestDriver testDriver;
37 TestInputTopic<TestInputUser, TestInputWord> in;
38 TestOutputTopic<TestOutputWord, TestOutputWordCounter> out;
42 public void setUpTestDriver()
44 Topology topology = CounterStreamProcessor.buildTopology(
47 Stores.inMemoryKeyValueStore(STORE_NAME));
49 testDriver = new TopologyTestDriver(topology, serializationConfig());
51 in = testDriver.createInputTopic(IN, serializer(), serializer());
52 out = testDriver.createOutputTopic(OUT, keyDeserializer(), valueDeserializer());
61 .forEach(kv -> in.pipeInput(kv.key, kv.value));
63 MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
66 .forEach(record -> receivedMessages.add(record.key(), record.value()));
68 TestData.assertExpectedMessages(receivedMessages);
70 TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
71 TestData.assertExpectedLastMessagesForWord(receivedMessages);
73 KeyValueStore<Word, Long> store = testDriver.getKeyValueStore(STORE_NAME);
74 TestData.assertExpectedState(store);
78 public void tearDown()
84 private static JsonSerializer serializer()
86 return new JsonSerializer().noTypeInfo();
89 private JsonDeserializer<TestOutputWord> keyDeserializer()
91 return deserializer(true);
94 private static JsonDeserializer<TestOutputWordCounter> valueDeserializer()
96 return deserializer(false);
99 private static <T> JsonDeserializer<T> deserializer(boolean isKey)
101 JsonDeserializer<T> deserializer = new JsonDeserializer<>();
102 deserializer.configure(
103 Map.of(JsonDeserializer.TYPE_MAPPINGS, typeMappingsConfig()),
108 private static String typeMappingsConfig()
110 return typeMappings()
113 .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
114 .collect(Collectors.joining(","));
117 private static Map<String, Class> typeMappings()
120 "word", TestOutputWord.class,
121 "counter", TestOutputWordCounter.class);