1 package de.juplo.kafka.wordcount.top10;
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.streams.*;
5 import org.apache.kafka.streams.state.Stores;
6 import org.junit.jupiter.api.Test;
7 import org.springframework.kafka.support.serializer.JsonDeserializer;
8 import org.springframework.kafka.support.serializer.JsonSerde;
9 import org.springframework.kafka.support.serializer.JsonSerializer;
11 import java.util.List;
13 import java.util.Properties;
15 import static de.juplo.kafka.wordcount.counter.TestData.convertToMap;
16 import static de.juplo.kafka.wordcount.counter.TestData.parseHeader;
17 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
18 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
22 public class Top10StreamProcessorTopologyTest
24 public final static String IN = "TEST-IN";
25 public final static String OUT = "TEST-OUT";
30 Topology topology = Top10StreamProcessor.buildTopology(IN, OUT);
32 Top10ApplicationConfiguriation applicationConfiguriation =
33 new Top10ApplicationConfiguriation();
34 Properties streamProcessorProperties =
35 applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
36 Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
38 JsonSerde<?> keySerde = new JsonSerde<>();
39 keySerde.configure(propertyMap, true);
40 JsonSerde<?> valueSerde = new JsonSerde<>();
41 valueSerde.configure(propertyMap, false);
43 TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
45 TestInputTopic<String, Word> in = testDriver.createInputTopic(
47 (JsonSerializer<String>)keySerde.serializer(),
48 (JsonSerializer<Word>)valueSerde.serializer());
50 TestOutputTopic<Word, WordCounter> out = testDriver.createOutputTopic(
52 (JsonDeserializer<Word>)keySerde.deserializer(),
53 (JsonDeserializer<WordCounter>)valueSerde.deserializer());
55 TestData.writeInputData((key, value) -> in.pipeInput(key, value));
57 List<KeyValue<Word, WordCounter>> receivedMessages = out
63 "OUT: {} -> {}, {}, {}",
66 parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
67 parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
68 return KeyValue.pair(record.key(), record.value());
72 TestData.assertExpectedResult(receivedMessages);