From: Kai Moritz Date: Sun, 26 May 2024 09:49:57 +0000 (+0200) Subject: splitter: 1.2.0 - Renamed test into `SplitterApplicationIT` -- MOVE X-Git-Tag: splitter-1.2.0~15 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=5dade68933e284e71667e0bea4ac578d2ec3f1d6;p=demos%2Fkafka%2Fwordcount splitter: 1.2.0 - Renamed test into `SplitterApplicationIT` -- MOVE --- diff --git a/src/test/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationIT.java b/src/test/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationIT.java new file mode 100644 index 0000000..3fa5851 --- /dev/null +++ b/src/test/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationIT.java @@ -0,0 +1,114 @@ +package de.juplo.kafka.wordcount.splitter; + +import com.fasterxml.jackson.databind.ObjectMapper; +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.splitter.SplitterApplicationTests.*; +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.splitter.bootstrap-server=${spring.embedded.kafka.brokers}", + "juplo.wordcount.splitter.input-topic=" + TOPIC_IN, + "juplo.wordcount.splitter.output-topic=" + TOPIC_OUT }) +@EmbeddedKafka(topics = { TOPIC_IN, TOPIC_OUT }, partitions = PARTITIONS) +@Slf4j +public class SplitterApplicationTests +{ + 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 + { + Recording recording = new Recording(); + recording.setUser("peter"); + recording.setSentence("Hallo Welt!"); + kafkaTemplate.send(TOPIC_IN, recording.getUser(), mapper.writeValueAsString(recording)); + recording.setUser("klaus"); + recording.setSentence("Müsch gäb's auch!"); + kafkaTemplate.send(TOPIC_IN, recording.getUser(), mapper.writeValueAsString(recording)); + recording.setUser("peter"); + recording.setSentence("Boäh, echt! ß mal nä Nümmäh!"); + kafkaTemplate.send(TOPIC_IN, recording.getUser(), mapper.writeValueAsString(recording)); + + String peter1 = mapper.writeValueAsString(Word.of("peter", "Hallo")); + String peter2 = mapper.writeValueAsString(Word.of("peter", "Welt")); + String peter3 = mapper.writeValueAsString(Word.of("peter", "Boäh")); + String peter4 = mapper.writeValueAsString(Word.of("peter", "echt")); + String peter5 = mapper.writeValueAsString(Word.of("peter", "ß")); + String peter6 = mapper.writeValueAsString(Word.of("peter", "mal")); + String peter7 = mapper.writeValueAsString(Word.of("peter", "nä")); + String peter8 = mapper.writeValueAsString(Word.of("peter", "Nümmäh")); + + String klaus1 = mapper.writeValueAsString(Word.of("klaus","Müsch")); + String klaus2 = mapper.writeValueAsString(Word.of("klaus","gäb")); + String klaus3 = mapper.writeValueAsString(Word.of("klaus","s")); + String klaus4 = mapper.writeValueAsString(Word.of("klaus","auch")); + + await("Expexted converted data") + .atMost(Duration.ofSeconds(5)) + .untilAsserted(() -> + { + assertThat(consumer.received).hasSize(2); + assertThat(consumer.received.get("klaus")).containsExactly(klaus1, klaus2, klaus3, klaus4); + assertThat(consumer.received.get("peter")).containsExactly(peter1, peter2, peter3, peter4, peter5, peter6, peter7, peter8); + }); + } + + + static class Consumer + { + final MultiValueMap received = new LinkedMultiValueMap<>(); + + @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) + public void receive(ConsumerRecord record) + { + log.debug("Received message: {}", record); + received.add(record.key(), record.value()); + } + } + + @TestConfiguration + static class Configuration + { + @Bean + Consumer consumer() + { + return new Consumer(); + } + } +} diff --git a/src/test/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationTests.java b/src/test/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationTests.java deleted file mode 100644 index 3fa5851..0000000 --- a/src/test/java/de/juplo/kafka/wordcount/splitter/SplitterApplicationTests.java +++ /dev/null @@ -1,114 +0,0 @@ -package de.juplo.kafka.wordcount.splitter; - -import com.fasterxml.jackson.databind.ObjectMapper; -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.splitter.SplitterApplicationTests.*; -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.splitter.bootstrap-server=${spring.embedded.kafka.brokers}", - "juplo.wordcount.splitter.input-topic=" + TOPIC_IN, - "juplo.wordcount.splitter.output-topic=" + TOPIC_OUT }) -@EmbeddedKafka(topics = { TOPIC_IN, TOPIC_OUT }, partitions = PARTITIONS) -@Slf4j -public class SplitterApplicationTests -{ - 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 - { - Recording recording = new Recording(); - recording.setUser("peter"); - recording.setSentence("Hallo Welt!"); - kafkaTemplate.send(TOPIC_IN, recording.getUser(), mapper.writeValueAsString(recording)); - recording.setUser("klaus"); - recording.setSentence("Müsch gäb's auch!"); - kafkaTemplate.send(TOPIC_IN, recording.getUser(), mapper.writeValueAsString(recording)); - recording.setUser("peter"); - recording.setSentence("Boäh, echt! ß mal nä Nümmäh!"); - kafkaTemplate.send(TOPIC_IN, recording.getUser(), mapper.writeValueAsString(recording)); - - String peter1 = mapper.writeValueAsString(Word.of("peter", "Hallo")); - String peter2 = mapper.writeValueAsString(Word.of("peter", "Welt")); - String peter3 = mapper.writeValueAsString(Word.of("peter", "Boäh")); - String peter4 = mapper.writeValueAsString(Word.of("peter", "echt")); - String peter5 = mapper.writeValueAsString(Word.of("peter", "ß")); - String peter6 = mapper.writeValueAsString(Word.of("peter", "mal")); - String peter7 = mapper.writeValueAsString(Word.of("peter", "nä")); - String peter8 = mapper.writeValueAsString(Word.of("peter", "Nümmäh")); - - String klaus1 = mapper.writeValueAsString(Word.of("klaus","Müsch")); - String klaus2 = mapper.writeValueAsString(Word.of("klaus","gäb")); - String klaus3 = mapper.writeValueAsString(Word.of("klaus","s")); - String klaus4 = mapper.writeValueAsString(Word.of("klaus","auch")); - - await("Expexted converted data") - .atMost(Duration.ofSeconds(5)) - .untilAsserted(() -> - { - assertThat(consumer.received).hasSize(2); - assertThat(consumer.received.get("klaus")).containsExactly(klaus1, klaus2, klaus3, klaus4); - assertThat(consumer.received.get("peter")).containsExactly(peter1, peter2, peter3, peter4, peter5, peter6, peter7, peter8); - }); - } - - - static class Consumer - { - final MultiValueMap received = new LinkedMultiValueMap<>(); - - @KafkaListener(groupId = "TEST", topics = TOPIC_OUT) - public void receive(ConsumerRecord record) - { - log.debug("Received message: {}", record); - received.add(record.key(), record.value()); - } - } - - @TestConfiguration - static class Configuration - { - @Bean - Consumer consumer() - { - return new Consumer(); - } - } -}