From f9106e5166ce84caa2a910aa21fb41efaebcc73f Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 12 May 2024 18:06:09 +0200 Subject: [PATCH] counter: 1.2.8 - Reconfigured tests to send data as domain-instances --- pom.xml | 2 +- .../counter/CounterApplicationIT.java | 31 ++++++++++++++++++- .../CounterStreamProcessorTopologyTest.java | 7 +++-- .../kafka/wordcount/counter/TestData.java | 24 +++++++------- 4 files changed, 47 insertions(+), 17 deletions(-) diff --git a/pom.xml b/pom.xml index 22b9da0..0b134f3 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount counter - 1.2.7 + 1.2.8 Wordcount-Counter Word-counting stream-processor of the multi-user wordcount-example diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java index b412fe4..559b171 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -3,6 +3,7 @@ package de.juplo.kafka.wordcount.counter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.Stores; import org.junit.jupiter.api.BeforeEach; @@ -13,12 +14,18 @@ import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Primary; import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.context.EmbeddedKafka; import java.time.Duration; import java.util.LinkedList; import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.*; import static org.awaitility.Awaitility.*; @@ -41,7 +48,7 @@ public class CounterApplicationIT static final int PARTITIONS = 2; @Autowired - KafkaTemplate kafkaTemplate; + KafkaTemplate kafkaTemplate; @Autowired Consumer consumer; @@ -85,6 +92,28 @@ public class CounterApplicationIT @TestConfiguration static class Configuration { + @Bean + ProducerFactory producerFactory(Properties streamProcessorProperties) + { + Map propertyMap = streamProcessorProperties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + + propertyMap.put( + ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, + JsonSerializer.class.getName()); + propertyMap.put( + ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + JsonSerializer.class.getName()); + + return new DefaultKafkaProducerFactory<>(propertyMap); + } + @Bean Consumer consumer() { 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 fe85dc4..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; @@ -33,10 +34,10 @@ public class CounterStreamProcessorTopologyTest 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, diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java index 8ff7022..f795e77 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -8,41 +8,41 @@ import static org.assertj.core.api.Assertions.assertThat; class TestData { - static void writeInputData(BiConsumer consumer) + static void writeInputData(BiConsumer consumer) { consumer.accept( "peter", - "{\"user\":\"peter\",\"word\":\"Hallo\"}"); + Word.of("peter","Hallo")); consumer.accept( "klaus", - "{\"user\":\"klaus\",\"word\":\"Müsch\"}"); + Word.of("klaus","Müsch")); consumer.accept( "peter", - "{\"user\":\"peter\",\"word\":\"Welt\"}"); + Word.of("peter","Welt")); consumer.accept( "klaus", - "{\"user\":\"klaus\",\"word\":\"Müsch\"}"); + Word.of("klaus","Müsch")); consumer.accept( "klaus", - "{\"user\":\"klaus\",\"word\":\"s\"}"); + Word.of("klaus","s")); consumer.accept( "peter", - "{\"user\":\"peter\",\"word\":\"Boäh\"}"); + Word.of("peter","Boäh")); consumer.accept( "peter", - "{\"user\":\"peter\",\"word\":\"Welt\"}"); + Word.of("peter","Welt")); consumer.accept( "peter", - "{\"user\":\"peter\",\"word\":\"Boäh\"}"); + Word.of("peter","Boäh")); consumer.accept( "klaus", - "{\"user\":\"klaus\",\"word\":\"s\"}"); + Word.of("klaus","s")); consumer.accept( "peter", - "{\"user\":\"peter\",\"word\":\"Boäh\"}"); + Word.of("peter","Boäh")); consumer.accept( "klaus", - "{\"user\":\"klaus\",\"word\":\"s\"}"); + Word.of("klaus","s")); } static void assertExpectedResult(List receivedMessages) -- 2.20.1