1 package de.juplo.kafka.wordcount.popular;
3 import de.juplo.kafka.wordcount.splitter.InputUser;
4 import de.juplo.kafka.wordcount.splitter.InputWord;
5 import de.juplo.kafka.wordcount.stats.OutputWindowedWord;
6 import de.juplo.kafka.wordcount.stats.OutputWordCounter;
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.AfterAll;
15 import org.junit.jupiter.api.BeforeAll;
16 import org.junit.jupiter.api.DisplayName;
17 import org.junit.jupiter.api.Test;
18 import org.springframework.kafka.support.serializer.JsonDeserializer;
19 import org.springframework.kafka.support.serializer.JsonSerializer;
20 import org.springframework.util.LinkedMultiValueMap;
21 import org.springframework.util.MultiValueMap;
23 import java.time.Duration;
24 import java.time.ZoneId;
27 import static de.juplo.kafka.wordcount.popular.PopularApplicationConfiguriation.serializationConfig;
28 import static de.juplo.kafka.wordcount.popular.PopularStreamProcessor.*;
32 public class PopularStreamProcessorTopologyTest
34 public static final String IN = "TEST-IN";
35 public static final String OUT = "TEST-OUT";
36 public static final ZoneId ZONE = ZoneId.of("Europe/Berlin");
39 static TopologyTestDriver testDriver;
40 static MultiValueMap<OutputWindowedWord, OutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
44 public static void setUpTestDriver()
46 Topology topology = PopularStreamProcessor.buildTopology(
50 Stores.inMemoryWindowStore(
52 WINDOW_SIZE.multipliedBy(2),
55 Stores.inMemoryKeyValueStore(KEY_VALUE_STORE_NAME));
57 testDriver = new TopologyTestDriver(topology, serializationConfig());
59 TestInputTopic<InputUser, InputWord> in =
60 testDriver.createInputTopic(IN, serializer(), serializer());
61 TestOutputTopic<OutputWindowedWord, OutputWordCounter> out =
62 testDriver.createOutputTopic(OUT, keyDeserializer(), valueDeserializer());
64 TestData.sendInputMessages((instant, kv) -> in.pipeInput(kv.key, kv.value, instant));
66 receivedMessages = new LinkedMultiValueMap<>();
69 .forEach(record -> receivedMessages.add(record.key(), record.value()));
73 @DisplayName("Assert the expected output messages")
75 public void testExpectedMessages()
77 TestData.assertExpectedMessages(receivedMessages);
80 @DisplayName("Assert the expected number of messages")
82 public void testExpectedNumberOfMessagesForWord()
84 TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
87 @DisplayName("Await the expected final output messages")
89 public void testExpectedLastMessagesForWord()
91 TestData.assertExpectedLastMessagesForWord(receivedMessages);
94 @DisplayName("Assert the expected state in the state-store")
96 public void testExpectedState()
98 KeyValueStore<WindowedWord, WordCounter> store = testDriver.getKeyValueStore(KEY_VALUE_STORE_NAME);
99 TestData.assertExpectedState(store);
103 public static void tearDown()
109 private static JsonSerializer serializer()
111 return new JsonSerializer().noTypeInfo();
114 private static JsonDeserializer<OutputWindowedWord> keyDeserializer()
116 return deserializer(true);
119 private static JsonDeserializer<OutputWordCounter> valueDeserializer()
121 return deserializer(false);
124 private static <T> JsonDeserializer<T> deserializer(boolean isKey)
126 JsonDeserializer<T> deserializer = new JsonDeserializer<>();
127 deserializer.configure(
128 Map.of(JsonDeserializer.TYPE_MAPPINGS, typeMappingsConfig()),
133 private static String typeMappingsConfig()
135 return PopularStreamProcessor.typeMappingsConfig(OutputWindowedWord.class, OutputWordCounter.class);