From 8a56919f493faa0acc6eb939b9c78312a1418efa Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 11 Feb 2023 07:09:45 +0100 Subject: [PATCH] counter: 1.1.3 - Implemented an integration test for the old message-format --- pom.xml | 25 ++- .../wordcount/counter/CounterApplication.java | 3 + .../counter/CounterApplicationIT.java | 149 ++++++++++++++++++ src/test/resources/logback-test.xml | 5 + 4 files changed, 181 insertions(+), 1 deletion(-) create mode 100644 src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java create mode 100644 src/test/resources/logback-test.xml diff --git a/pom.xml b/pom.xml index 83f96cd..1d9425d 100644 --- a/pom.xml +++ b/pom.xml @@ -10,7 +10,7 @@ de.juplo.kafka.wordcount counter - 1.1.2 + 1.1.3 Wordcount-Counter Word-counting stream-processor of the multi-user wordcount-example @@ -29,6 +29,10 @@ org.apache.kafka kafka-streams + + org.springframework.kafka + spring-kafka + org.springframework.boot @@ -46,15 +50,34 @@ lombok true + org.springframework.boot spring-boot-starter-test test + + org.springframework.kafka + spring-kafka-test + test + + + org.awaitility + awaitility + test + + + org.assertj + assertj-core + test + + + maven-failsafe-plugin + org.springframework.boot spring-boot-maven-plugin diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplication.java b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplication.java index 1ee65ca..6a6a78c 100644 --- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplication.java +++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterApplication.java @@ -34,6 +34,9 @@ public class CounterApplication propertyMap.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer()); propertyMap.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName()); propertyMap.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName()); + propertyMap.put(StreamsConfig.STATE_DIR_CONFIG, "target"); + propertyMap.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 0); + propertyMap.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); propertyMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); CounterStreamProcessor streamProcessor = new CounterStreamProcessor( diff --git a/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java new file mode 100644 index 0000000..aad79b8 --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/counter/CounterApplicationIT.java @@ -0,0 +1,149 @@ +package de.juplo.kafka.wordcount.counter; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.RequiredArgsConstructor; +import lombok.Value; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; + +import java.time.Duration; + +import static de.juplo.kafka.wordcount.counter.CounterApplicationIT.*; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.*; + + +@SpringBootTest( + properties = { + "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}", + "juplo.wordcount.counter.bootstrap-server=${spring.embedded.kafka.brokers}", + "juplo.wordcount.counter.input-topic=" + TOPIC_IN, + "juplo.wordcount.counter.output-topic=" + TOPIC_OUT }) +@EmbeddedKafka(topics = { TOPIC_IN, TOPIC_OUT }, partitions = PARTITIONS) +@Slf4j +public class CounterApplicationIT +{ + public final static String TOPIC_IN = "in"; + public final static String TOPIC_OUT = "out"; + static final int PARTITIONS = 2; + + @Autowired + KafkaTemplate kafkaTemplate; + @Autowired + ObjectMapper mapper; + @Autowired + Consumer consumer; + + + @BeforeEach + public void clear() + { + consumer.received.clear(); + } + + + @Test + void testSendMessage() throws Exception + { + kafkaTemplate.send(TOPIC_IN, "peter", "Hallo"); + kafkaTemplate.send(TOPIC_IN, "klaus", "Müsch"); + kafkaTemplate.send(TOPIC_IN, "peter", "Welt"); + kafkaTemplate.send(TOPIC_IN, "klaus", "Müsch"); + kafkaTemplate.send(TOPIC_IN, "klaus", "s"); + kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); + kafkaTemplate.send(TOPIC_IN, "peter", "Welt"); + kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); + kafkaTemplate.send(TOPIC_IN, "klaus", "s"); + kafkaTemplate.send(TOPIC_IN, "peter", "Boäh"); + kafkaTemplate.send(TOPIC_IN, "klaus", "s"); + + Message peter1 = Message.of( + "{\"username\":\"peter\",\"word\":\"Hallo\"}", + "1"); + Message peter2 = Message.of( + "{\"username\":\"peter\",\"word\":\"Welt\"}", + "1"); + Message peter3 = Message.of( + "{\"username\":\"peter\",\"word\":\"Boäh\"}", + "1"); + Message peter4 = Message.of( + "{\"username\":\"peter\",\"word\":\"Welt\"}", + "2"); + Message peter5 = Message.of( + "{\"username\":\"peter\",\"word\":\"Boäh\"}", + "2"); + Message peter6 = Message.of( + "{\"username\":\"peter\",\"word\":\"Boäh\"}", + "3"); + + Message klaus1 = Message.of( + "{\"username\":\"klaus\",\"word\":\"Müsch\"}", + "1"); + Message klaus2 = Message.of( + "{\"username\":\"klaus\",\"word\":\"Müsch\"}", + "2"); + Message klaus3 = Message.of( + "{\"username\":\"klaus\",\"word\":\"s\"}", + "1"); + Message klaus4 = Message.of( + "{\"username\":\"klaus\",\"word\":\"s\"}", + "2"); + Message klaus5 = Message.of( + "{\"username\":\"klaus\",\"word\":\"s\"}", + "3"); + + await("Expexted converted data") + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> + { + assertThat(consumer.received).hasSize(2); + assertThat(consumer.received.get("klaus")).containsExactly(klaus1, klaus2, klaus3, klaus4, klaus5); + assertThat(consumer.received.get("peter")).containsExactly(peter1, peter2, peter3, peter4, peter5, peter6); + }); + } + + + @RequiredArgsConstructor + static class Consumer + { + private final MultiValueMap received = new LinkedMultiValueMap<>(); + private final ObjectMapper mapper; + + @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) + public void receive(ConsumerRecord record) throws JsonProcessingException + { + log.debug("Received message: {}", record); + Key key = mapper.readValue(record.key(), Key.class); + received.add(key.getUsername(), Message.of(record.key(),record.value())); + } + } + + @Value(staticConstructor = "of") + static class Message + { + String key; + String value; + } + + @TestConfiguration + static class Configuration + { + @Bean + Consumer consumer(ObjectMapper mapper) + { + return new Consumer(mapper); + } + } +} diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml new file mode 100644 index 0000000..171bf63 --- /dev/null +++ b/src/test/resources/logback-test.xml @@ -0,0 +1,5 @@ + + + + + -- 2.20.1