1 package de.juplo.kafka.wordcount.top10;
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.streams.*;
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;
14 import static de.juplo.kafka.wordcount.top10.TestData.convertToMap;
15 import static de.juplo.kafka.wordcount.top10.TestData.parseHeader;
16 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
17 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
21 public class Top10StreamProcessorTopologyTest
23 public final static String IN = "TEST-IN";
24 public final static String OUT = "TEST-OUT";
29 Topology topology = Top10StreamProcessor.buildTopology(IN, OUT);
31 Top10ApplicationConfiguration applicationConfiguriation =
32 new Top10ApplicationConfiguration();
33 Properties streamProcessorProperties =
34 applicationConfiguriation.streamProcessorProperties(new Top10ApplicationProperties());
35 Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
37 JsonSerde<?> keySerde = new JsonSerde<>();
38 keySerde.configure(propertyMap, true);
39 JsonSerde<?> valueSerde = new JsonSerde<>();
40 valueSerde.configure(propertyMap, false);
42 TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
44 TestInputTopic<Key, Entry> in = testDriver.createInputTopic(
46 (JsonSerializer<Key>)keySerde.serializer(),
47 (JsonSerializer<Entry>)valueSerde.serializer());
49 TestOutputTopic<String, Ranking> out = testDriver.createOutputTopic(
51 (JsonDeserializer<String>)keySerde.deserializer(),
52 (JsonDeserializer<Ranking>)valueSerde.deserializer());
54 TestData.writeInputData((key, value) -> in.pipeInput(
56 Entry.of(value.getWord(), value.getCounter())));
58 List<KeyValue<String, Ranking>> receivedMessages = out
64 "OUT: {} -> {}, {}, {}",
67 parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
68 parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
69 return KeyValue.pair(record.key(), record.value());
73 TestData.assertExpectedResult(receivedMessages);