1 package de.juplo.kafka.wordcount.counter;
3 import org.apache.kafka.streams.TestInputTopic;
4 import org.apache.kafka.streams.TestOutputTopic;
5 import org.apache.kafka.streams.Topology;
6 import org.apache.kafka.streams.TopologyTestDriver;
7 import org.apache.kafka.streams.state.Stores;
8 import org.junit.jupiter.api.Test;
9 import org.springframework.kafka.support.serializer.JsonDeserializer;
10 import org.springframework.kafka.support.serializer.JsonSerde;
11 import org.springframework.kafka.support.serializer.JsonSerializer;
13 import java.util.List;
15 import java.util.Properties;
16 import java.util.stream.Collectors;
19 public class CounterStreamProcessorTopologyTest
21 public final static String IN = "TEST-IN";
22 public final static String OUT = "TEST-OUT";
27 Topology topology = CounterStreamProcessor.buildTopology(
30 Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
32 CounterApplicationConfiguriation applicationConfiguriation =
33 new CounterApplicationConfiguriation();
34 Properties streamProcessorProperties =
35 applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
37 Map<String, ?> propertyMap = streamProcessorProperties
42 entry -> (String)entry.getKey(),
43 entry -> entry.getValue()
45 JsonSerde<?> keySerde = new JsonSerde<>();
46 keySerde.configure(propertyMap, true);
47 JsonSerde<?> valueSerde = new JsonSerde<>();
48 valueSerde.configure(propertyMap, false);
50 TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
52 TestInputTopic<String, Word> in = testDriver.createInputTopic(
54 (JsonSerializer<String>)keySerde.serializer(),
55 (JsonSerializer<Word>)valueSerde.serializer());
57 TestOutputTopic<Word, WordCount> out = testDriver.createOutputTopic(
59 (JsonDeserializer<Word>)keySerde.deserializer(),
60 (JsonDeserializer<WordCount>)valueSerde.deserializer());
62 TestData.writeInputData((key, value) -> in.pipeInput(key, value));
64 List<Message> receivedMessages = out
67 .map(record -> Message.of(record.key(), record.value()))
70 TestData.assertExpectedResult(receivedMessages);