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;
22 import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
26 public class CounterStreamProcessorTopologyTest
28 public static final String IN = "TEST-IN";
29 public static final String OUT = "TEST-OUT";
30 public static final String STORE_NAME = "TOPOLOGY-TEST";
33 TopologyTestDriver testDriver;
34 TestInputTopic<TestInputUser, TestInputWord> in;
35 TestOutputTopic<TestOutputWord, TestOutputWordCounter> out;
39 public void setUpTestDriver()
41 Topology topology = CounterStreamProcessor.buildTopology(
44 Stores.inMemoryKeyValueStore(STORE_NAME));
46 testDriver = new TopologyTestDriver(topology, serializationConfig());
48 in = testDriver.createInputTopic(
50 new JsonSerializer().noTypeInfo(),
51 new JsonSerializer().noTypeInfo());
53 out = testDriver.createOutputTopic(
55 new JsonDeserializer()
56 .copyWithType(TestOutputWord.class)
58 new JsonDeserializer()
59 .copyWithType(TestOutputWordCounter.class)
60 .ignoreTypeHeaders());
69 .forEach(kv -> in.pipeInput(kv.key, kv.value));
71 MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
74 .forEach(record -> receivedMessages.add(record.key(), record.value()));
76 TestData.assertExpectedMessages(receivedMessages);
78 TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
79 TestData.assertExpectedLastMessagesForWord(receivedMessages);
81 KeyValueStore<Word, Long> store = testDriver.getKeyValueStore(STORE_NAME);
82 TestData.assertExpectedState(store);
86 public void tearDown()