1 package de.juplo.kafka.wordcount.top10;
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.streams.TestInputTopic;
5 import org.apache.kafka.streams.TestOutputTopic;
6 import org.apache.kafka.streams.Topology;
7 import org.apache.kafka.streams.TopologyTestDriver;
8 import org.apache.kafka.streams.state.KeyValueStore;
9 import org.apache.kafka.streams.state.Stores;
10 import org.junit.jupiter.api.AfterEach;
11 import org.junit.jupiter.api.BeforeEach;
12 import org.junit.jupiter.api.Test;
13 import org.springframework.kafka.support.serializer.JsonDeserializer;
14 import org.springframework.kafka.support.serializer.JsonSerde;
15 import org.springframework.kafka.support.serializer.JsonSerializer;
16 import org.springframework.util.LinkedMultiValueMap;
17 import org.springframework.util.MultiValueMap;
20 import java.util.Properties;
21 import java.util.stream.Stream;
23 import static de.juplo.kafka.wordcount.top10.Top10ApplicationConfiguration.serializationConfig;
27 public class Top10StreamProcessorTopologyTest
29 public static final String IN = "TEST-IN";
30 public static final String OUT = "TEST-OUT";
31 public static final String STORE_NAME = "TOPOLOGY-TEST";
34 TopologyTestDriver testDriver;
35 TestInputTopic<Key, Entry> in;
36 TestOutputTopic<User, Ranking> out;
42 Topology topology = Top10StreamProcessor.buildTopology(
45 Stores.inMemoryKeyValueStore(STORE_NAME));
47 Map<String, Object> propertyMap = serializationConfig();
49 Properties properties = new Properties();
50 properties.putAll(propertyMap);
52 JsonSerde<?> keySerde = new JsonSerde<>();
53 keySerde.configure(propertyMap, true);
54 JsonSerde<?> valueSerde = new JsonSerde<>();
55 valueSerde.configure(propertyMap, false);
57 testDriver = new TopologyTestDriver(topology, properties);
59 in = testDriver.createInputTopic(
61 (JsonSerializer<Key>)keySerde.serializer(),
62 (JsonSerializer<Entry>)valueSerde.serializer());
64 out = testDriver.createOutputTopic(
66 (JsonDeserializer<User>)keySerde.deserializer(),
67 (JsonDeserializer<Ranking>)valueSerde.deserializer());
76 .of(TestData.INPUT_MESSAGES)
77 .forEach(kv -> in.pipeInput(
78 Key.of(kv.key.getUser(), kv.key.getWord()),
79 Entry.of(kv.value.getWord(), kv.value.getCounter())));
81 MultiValueMap<User, Ranking> receivedMessages = new LinkedMultiValueMap<>();
84 .forEach(record -> receivedMessages.add(record.key(), record.value()));
86 TestData.assertExpectedMessages(receivedMessages);
88 TestData.assertExpectedNumberOfMessagesForUsers(receivedMessages);
89 TestData.assertExpectedLastMessagesForUsers(receivedMessages);
91 KeyValueStore<User, Ranking> store = testDriver.getKeyValueStore(STORE_NAME);
92 TestData.assertExpectedState(store);
96 public void tearDown()