From: Kai Moritz Date: Tue, 14 May 2024 21:48:34 +0000 (+0200) Subject: WIP X-Git-Url: https://juplo.de/gitweb/?a=commitdiff_plain;h=e7c52491bcef70efd85487b7469a32e832d61724;p=demos%2Fkafka%2Fwordcount WIP --- diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java new file mode 100644 index 0000000..8b4a593 --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java @@ -0,0 +1,77 @@ +package de.juplo.kafka.wordcount.top10; + +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.streams.*; +import org.apache.kafka.streams.state.Stores; +import org.junit.jupiter.api.Test; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.support.serializer.JsonSerializer; + +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import static de.juplo.kafka.wordcount.counter.TestData.convertToMap; +import static de.juplo.kafka.wordcount.counter.TestData.parseHeader; +import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME; +import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME; + + +@Slf4j +public class Top10StreamProcessorTopologyTest +{ + public final static String IN = "TEST-IN"; + public final static String OUT = "TEST-OUT"; + + @Test + public void test() + { + Topology topology = CounterStreamProcessor.buildTopology( + IN, + OUT, + Stores.inMemoryKeyValueStore("TOPOLOGY-TEST")); + + CounterApplicationConfiguriation applicationConfiguriation = + new CounterApplicationConfiguriation(); + Properties streamProcessorProperties = + applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties()); + Map propertyMap = convertToMap(streamProcessorProperties); + + JsonSerde keySerde = new JsonSerde<>(); + keySerde.configure(propertyMap, true); + JsonSerde valueSerde = new JsonSerde<>(); + valueSerde.configure(propertyMap, false); + + TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties); + + TestInputTopic in = testDriver.createInputTopic( + IN, + (JsonSerializer)keySerde.serializer(), + (JsonSerializer)valueSerde.serializer()); + + TestOutputTopic out = testDriver.createOutputTopic( + OUT, + (JsonDeserializer)keySerde.deserializer(), + (JsonDeserializer)valueSerde.deserializer()); + + TestData.writeInputData((key, value) -> in.pipeInput(key, value)); + + List> receivedMessages = out + .readRecordsToList() + .stream() + .map(record -> + { + log.debug( + "OUT: {} -> {}, {}, {}", + record.key(), + record.value(), + parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME), + parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME)); + return KeyValue.pair(record.key(), record.value()); + }) + .toList(); + + TestData.assertExpectedResult(receivedMessages); + } +}