1 package de.juplo.kafka.wordcount.counter;
3 import org.apache.kafka.streams.*;
4 import org.apache.kafka.streams.state.Stores;
5 import org.junit.jupiter.api.Test;
6 import org.springframework.kafka.support.serializer.JsonDeserializer;
7 import org.springframework.kafka.support.serializer.JsonSerde;
8 import org.springframework.kafka.support.serializer.JsonSerializer;
10 import java.util.List;
12 import java.util.Properties;
13 import java.util.stream.Collectors;
16 public class CounterStreamProcessorTopologyTest
18 public final static String IN = "TEST-IN";
19 public final static String OUT = "TEST-OUT";
24 Topology topology = CounterStreamProcessor.buildTopology(
27 Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
29 CounterApplicationConfiguriation applicationConfiguriation =
30 new CounterApplicationConfiguriation();
31 Properties streamProcessorProperties =
32 applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
34 Map<String, ?> propertyMap = streamProcessorProperties
39 entry -> (String)entry.getKey(),
40 entry -> entry.getValue()
42 JsonSerde<?> keySerde = new JsonSerde<>();
43 keySerde.configure(propertyMap, true);
44 JsonSerde<?> valueSerde = new JsonSerde<>();
45 valueSerde.configure(propertyMap, false);
47 TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
49 TestInputTopic<String, Word> in = testDriver.createInputTopic(
51 (JsonSerializer<String>)keySerde.serializer(),
52 (JsonSerializer<Word>)valueSerde.serializer());
54 TestOutputTopic<Word, WordCount> out = testDriver.createOutputTopic(
56 (JsonDeserializer<Word>)keySerde.deserializer(),
57 (JsonDeserializer<WordCount>)valueSerde.deserializer());
59 TestData.writeInputData((key, value) -> in.pipeInput(key, value));
61 List<KeyValue<Word,WordCount>> receivedMessages = out
64 .map(record -> KeyValue.pair(record.key(), record.value()))
67 TestData.assertExpectedResult(receivedMessages);