1 package de.juplo.kafka.wordcount.counter;
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.Stores;
9 import org.junit.jupiter.api.AfterEach;
10 import org.junit.jupiter.api.BeforeEach;
11 import org.junit.jupiter.api.Test;
12 import org.springframework.kafka.support.serializer.JsonDeserializer;
13 import org.springframework.kafka.support.serializer.JsonSerde;
14 import org.springframework.kafka.support.serializer.JsonSerializer;
15 import org.springframework.util.LinkedMultiValueMap;
16 import org.springframework.util.MultiValueMap;
19 import java.util.Properties;
20 import java.util.stream.Stream;
22 import static de.juplo.kafka.wordcount.counter.TestData.convertToMap;
23 import static de.juplo.kafka.wordcount.counter.TestData.parseHeader;
24 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
25 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
29 public class CounterStreamProcessorTopologyTest
31 public final static String IN = "TEST-IN";
32 public final static String OUT = "TEST-OUT";
35 TopologyTestDriver testDriver;
36 TestInputTopic<String, Word> in;
37 TestOutputTopic<Word, WordCounter> out;
41 public void setUpTestDriver()
43 Topology topology = CounterStreamProcessor.buildTopology(
46 Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
48 CounterApplicationConfiguriation applicationConfiguriation =
49 new CounterApplicationConfiguriation();
50 Properties streamProcessorProperties =
51 applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
52 Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
54 JsonSerde<?> keySerde = new JsonSerde<>();
55 keySerde.configure(propertyMap, true);
56 JsonSerde<?> valueSerde = new JsonSerde<>();
57 valueSerde.configure(propertyMap, false);
59 testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
61 in = testDriver.createInputTopic(
63 (JsonSerializer<String>)keySerde.serializer(),
64 (JsonSerializer<Word>)valueSerde.serializer());
66 out = testDriver.createOutputTopic(
68 (JsonDeserializer<Word>)keySerde.deserializer(),
69 (JsonDeserializer<WordCounter>)valueSerde.deserializer());
77 .of(TestData.INPUT_MESSAGES)
78 .forEach(word -> in.pipeInput(word.getUser(), word));
80 MultiValueMap<Word, WordCounter> receivedMessages = new LinkedMultiValueMap<>();
86 "OUT: {} -> {}, {}, {}",
89 parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
90 parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
91 receivedMessages.add(record.key(), record.value());
94 TestData.assertExpectedMessages(receivedMessages);
98 public void tearDown()