1 package de.juplo.kafka.wordcount.counter;
3 import org.apache.kafka.common.serialization.StringDeserializer;
4 import org.apache.kafka.common.serialization.StringSerializer;
5 import org.apache.kafka.streams.TestInputTopic;
6 import org.apache.kafka.streams.TestOutputTopic;
7 import org.apache.kafka.streams.Topology;
8 import org.apache.kafka.streams.TopologyTestDriver;
9 import org.apache.kafka.streams.state.Stores;
10 import org.junit.jupiter.api.Test;
11 import org.springframework.kafka.support.serializer.JsonSerializer;
13 import java.util.List;
14 import java.util.Properties;
17 public class CounterStreamProcessorTopologyTest
19 public final static String IN = "TEST-IN";
20 public final static String OUT = "TEST-OUT";
25 Topology topology = CounterStreamProcessor.buildTopology(
28 Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
30 CounterApplicationConfiguriation applicationConfiguriation =
31 new CounterApplicationConfiguriation();
32 Properties streamProcessorProperties =
33 applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
35 TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
37 TestInputTopic<String, Word> in = testDriver.createInputTopic(
39 new JsonSerializer<>(),
40 new JsonSerializer<>());
42 TestOutputTopic<String, String> out = testDriver.createOutputTopic(
44 new StringDeserializer(),
45 new StringDeserializer());
47 TestData.writeInputData((key, value) -> in.pipeInput(key, value));
49 List<Message> receivedMessages = out
52 .map(record -> Message.of(record.key(), record.value()))
55 TestData.assertExpectedResult(receivedMessages);