From a7c3a493ee3b6ec8629eb749d1ce4ed059f4cff9 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 8 Jun 2024 18:05:45 +0200 Subject: [PATCH] top10: 1.2.1 - Simplified setup of `Top10StreamProcessorToplogyTest` --- .../top10/Top10ApplicationConfiguration.java | 6 +-- .../Top10StreamProcessorTopologyTest.java | 40 +++++++++++-------- 2 files changed, 25 insertions(+), 21 deletions(-) diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java b/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java index bd5298d..5e56066 100644 --- a/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/wordcount/top10/Top10ApplicationConfiguration.java @@ -13,8 +13,6 @@ import org.springframework.context.annotation.Configuration; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerde; -import java.util.HashMap; -import java.util.Map; import java.util.Properties; import java.util.concurrent.CompletableFuture; @@ -46,9 +44,9 @@ public class Top10ApplicationConfiguration return props; } - static Map serializationConfig() + static Properties serializationConfig() { - Map props = new HashMap<>(); + Properties props = new Properties(); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); diff --git a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java index 84cfb1e..80fc0df 100644 --- a/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java @@ -1,5 +1,7 @@ package de.juplo.kafka.wordcount.top10; +import de.juplo.kafka.wordcount.counter.TestCounter; +import de.juplo.kafka.wordcount.counter.TestWord; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.streams.TestInputTopic; import org.apache.kafka.streams.TestOutputTopic; @@ -11,13 +13,11 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; 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 org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import java.util.Map; -import java.util.Properties; import java.util.stream.Stream; import static de.juplo.kafka.wordcount.top10.Top10ApplicationConfiguration.serializationConfig; @@ -44,27 +44,21 @@ public class Top10StreamProcessorTopologyTest OUT, Stores.inMemoryKeyValueStore(STORE_NAME)); - Map propertyMap = serializationConfig(); - - Properties properties = new Properties(); - properties.putAll(propertyMap); - - JsonSerde keySerde = new JsonSerde<>(); - keySerde.configure(propertyMap, true); - JsonSerde valueSerde = new JsonSerde<>(); - valueSerde.configure(propertyMap, false); - - testDriver = new TopologyTestDriver(topology, properties); + testDriver = new TopologyTestDriver(topology, serializationConfig()); in = testDriver.createInputTopic( IN, - (JsonSerializer)keySerde.serializer(), - (JsonSerializer)valueSerde.serializer()); + jsonSerializer(Key.class, true), + jsonSerializer(Entry.class,false)); out = testDriver.createOutputTopic( OUT, - (JsonDeserializer)keySerde.deserializer(), - (JsonDeserializer)valueSerde.deserializer()); + new JsonDeserializer() + .copyWithType(User.class) + .ignoreTypeHeaders(), + new JsonDeserializer() + .copyWithType(Ranking.class) + .ignoreTypeHeaders()); } @@ -97,4 +91,16 @@ public class Top10StreamProcessorTopologyTest { testDriver.close(); } + + private JsonSerializer jsonSerializer(Class type, boolean isKey) + { + JsonSerializer jsonSerializer = new JsonSerializer<>(); + jsonSerializer.configure( + Map.of( + JsonSerializer.TYPE_MAPPINGS, + "word:" + TestWord.class.getName() + "," + + "counter:" + TestCounter.class.getName()), + isKey); + return jsonSerializer; + } } -- 2.20.1