From ff29d17cc2b1fa9b465c1874e387c23b7d7adc4e Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 12 May 2024 23:30:04 +0200 Subject: [PATCH] counter: 1.2.9 - Reconfigured tests to receive data as domain-instances --- pom.xml | 2 +- .../CounterApplicationConfiguriation.java | 3 -- .../kafka/wordcount/counter/WordCount.java | 8 +++- .../counter/CounterApplicationIT.java | 40 +++++++++++++++-- .../CounterStreamProcessorTopologyTest.java | 29 +++++++++--- .../kafka/wordcount/counter/Message.java | 4 +- .../kafka/wordcount/counter/TestData.java | 44 +++++++++---------- 7 files changed, 89 insertions(+), 41 deletions(-) diff --git a/pom.xml b/pom.xml index 0b134f3..a710d47 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount counter - 1.2.8 + 1.2.9 Wordcount-Counter Word-counting stream-processor of the multi-user wordcount-example diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java index 9785f69..f836f89 100644 --- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java +++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplicationConfiguriation.java @@ -12,7 +12,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerde; -import org.springframework.kafka.support.serializer.JsonSerializer; import java.util.Properties; import java.util.concurrent.CompletableFuture; @@ -34,11 +33,9 @@ public class CounterApplicationConfiguriation propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, counterProperties.getBootstrapServer()); propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); - propertyMap.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false); propertyMap.put(JsonDeserializer.TRUSTED_PACKAGES, CounterApplication.class.getPackageName()); propertyMap.put(JsonDeserializer.KEY_DEFAULT_TYPE, Word.class.getName()); propertyMap.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Word.class.getName()); - propertyMap.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false); propertyMap.put(StreamsConfig.STATE_DIR_CONFIG, "target"); if (counterProperties.getCommitInterval() != null) propertyMap.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, counterProperties.getCommitInterval()); diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java b/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java index 0767dd1..958f9b7 100644 --- a/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java +++ b/src/main/java/de/juplo/kafka/wordcount/counter/WordCount.java @@ -1,9 +1,13 @@ package de.juplo.kafka.wordcount.counter; -import lombok.Value; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; -@Value(staticConstructor = "of") +@Data +@NoArgsConstructor +@AllArgsConstructor(staticName = "of") public class WordCount { String user; 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 559b171..158c6ae 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -2,6 +2,7 @@ package de.juplo.kafka.wordcount.counter; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; @@ -14,9 +15,9 @@ 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.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.core.*; +import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.context.EmbeddedKafka; @@ -77,7 +78,7 @@ public class CounterApplicationIT private final List received = new LinkedList<>(); @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) - public synchronized void receive(ConsumerRecord record) + public synchronized void receive(ConsumerRecord record) { log.debug("Received message: {}", record); received.add(Message.of(record.key(),record.value())); @@ -114,6 +115,37 @@ public class CounterApplicationIT return new DefaultKafkaProducerFactory<>(propertyMap); } + @Bean + ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory( + Properties streamProcessorProperties) + { + Map propertyMap = streamProcessorProperties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + + propertyMap.put( + ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, + JsonDeserializer.class.getName()); + propertyMap.put( + ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, + JsonDeserializer.class.getName()); + + ConsumerFactory consumerFactory = + new DefaultKafkaConsumerFactory<>(propertyMap); + + ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory = + new ConcurrentKafkaListenerContainerFactory<>(); + + kafkaListenerContainerFactory.setConsumerFactory(consumerFactory); + + return kafkaListenerContainerFactory; + } + @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 902e93f..42ca78b 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java @@ -1,17 +1,19 @@ package de.juplo.kafka.wordcount.counter; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.streams.TestInputTopic; import org.apache.kafka.streams.TestOutputTopic; 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.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 java.util.stream.Collectors; public class CounterStreamProcessorTopologyTest @@ -32,17 +34,30 @@ public class CounterStreamProcessorTopologyTest Properties streamProcessorProperties = applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties()); + Map propertyMap = streamProcessorProperties + .entrySet() + .stream() + .collect( + Collectors.toMap( + entry -> (String)entry.getKey(), + entry -> entry.getValue() + )); + 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, - new JsonSerializer<>(), - new JsonSerializer<>()); + (JsonSerializer)keySerde.serializer(), + (JsonSerializer)valueSerde.serializer()); - TestOutputTopic out = testDriver.createOutputTopic( + TestOutputTopic out = testDriver.createOutputTopic( OUT, - new StringDeserializer(), - new StringDeserializer()); + (JsonDeserializer)keySerde.deserializer(), + (JsonDeserializer)valueSerde.deserializer()); TestData.writeInputData((key, value) -> in.pipeInput(key, value)); diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/Message.java b/src/test/java/de/juplo/kafka/wordcount/counter/Message.java index 8ed7cc5..eb64e6d 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/Message.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/Message.java @@ -6,6 +6,6 @@ import lombok.Value; @Value(staticConstructor = "of") public class Message { - String key; - String value; + Word key; + WordCount value; } 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 f795e77..4a65a78 100644 --- a/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java +++ b/src/test/java/de/juplo/kafka/wordcount/counter/TestData.java @@ -69,37 +69,37 @@ class TestData static Message[] expectedMessages = { Message.of( - "{\"user\":\"peter\",\"word\":\"Hallo\"}", - "{\"user\":\"peter\",\"word\":\"Hallo\",\"count\":1}"), + Word.of("peter","Hallo"), + WordCount.of("peter","Hallo",1)), Message.of( - "{\"user\":\"klaus\",\"word\":\"Müsch\"}", - "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":1}"), + Word.of("klaus","Müsch"), + WordCount.of("klaus","Müsch",1)), Message.of( - "{\"user\":\"peter\",\"word\":\"Welt\"}", - "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":1}"), + Word.of("peter","Welt"), + WordCount.of("peter","Welt",1)), Message.of( - "{\"user\":\"klaus\",\"word\":\"Müsch\"}", - "{\"user\":\"klaus\",\"word\":\"Müsch\",\"count\":2}"), + Word.of("klaus","Müsch"), + WordCount.of("klaus","Müsch",2)), Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":1}"), + Word.of("klaus","s"), + WordCount.of("klaus","s",1)), Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":1}"), + Word.of("peter","Boäh"), + WordCount.of("peter","Boäh",1)), Message.of( - "{\"user\":\"peter\",\"word\":\"Welt\"}", - "{\"user\":\"peter\",\"word\":\"Welt\",\"count\":2}"), + Word.of("peter","Welt"), + WordCount.of("peter","Welt",2)), Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":2}"), + Word.of("peter","Boäh"), + WordCount.of("peter","Boäh",2)), Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":2}"), + Word.of("klaus","s"), + WordCount.of("klaus","s",2)), Message.of( - "{\"user\":\"peter\",\"word\":\"Boäh\"}", - "{\"user\":\"peter\",\"word\":\"Boäh\",\"count\":3}"), + Word.of("peter","Boäh"), + WordCount.of("peter","Boäh",3)), Message.of( - "{\"user\":\"klaus\",\"word\":\"s\"}", - "{\"user\":\"klaus\",\"word\":\"s\",\"count\":3}"), + Word.of("klaus","s"), + WordCount.of("klaus","s",3)), }; } -- 2.20.1