X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterStreamProcessorTopologyTest.java;h=902e93f379566a8a973f915a63e3f26d1529f7be;hb=f9106e5166ce84caa2a910aa21fb41efaebcc73f;hp=2c8bd1e6c5e0d02db5949577a87dd4e3516aabfb;hpb=3ff17b238afc765fe944a20ef67a2f6288fd5df2;p=demos%2Fkafka%2Fwordcount diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java index 2c8bd1e..902e93f 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java @@ -8,6 +8,7 @@ import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.TopologyTestDriver; import org.apache.kafka.streams.state.Stores; import org.junit.jupiter.api.Test; +import org.springframework.kafka.support.serializer.JsonSerializer; import java.util.List; import java.util.Properties; @@ -26,17 +27,17 @@ public class CounterStreamProcessorTopologyTest OUT, Stores.inMemoryKeyValueStore("TOPOLOGY-TEST")); - CounterApplicationConfiguriation config = + CounterApplicationConfiguriation applicationConfiguriation = new CounterApplicationConfiguriation(); - Properties properties = - config.propertyMap(new CounterApplicationProperties()); + Properties streamProcessorProperties = + applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties()); - TopologyTestDriver testDriver = new TopologyTestDriver(topology, properties); + TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties); - TestInputTopic in = testDriver.createInputTopic( + TestInputTopic in = testDriver.createInputTopic( IN, - new StringSerializer(), - new StringSerializer()); + new JsonSerializer<>(), + new JsonSerializer<>()); TestOutputTopic out = testDriver.createOutputTopic( OUT,