X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fsplitter%2FApplicationTests.java;h=113501ffef40772cdd76fdc21c71e9b5ca03da46;hb=9c9ffbe3316ed295533c576e823794aa6de99665;hp=77266a94d30e7b43c66afe9da15131720ae2ea4c;hpb=d5f54354b2b44d125493c830bf0475f7992ee395;p=demos%2Fkafka%2Fwordcount

diff --git a/src/test/java/de/juplo/kafka/wordcount/splitter/ApplicationTests.java b/src/test/java/de/juplo/kafka/wordcount/splitter/ApplicationTests.java
index 77266a9..113501f 100644
--- a/src/test/java/de/juplo/kafka/wordcount/splitter/ApplicationTests.java
+++ b/src/test/java/de/juplo/kafka/wordcount/splitter/ApplicationTests.java
@@ -1,13 +1,104 @@
 package de.juplo.kafka.wordcount.splitter;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.awaitility.Awaitility;
+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;
 
-@SpringBootTest
+import java.time.Duration;
+import java.util.LinkedList;
+import java.util.List;
+
+import static de.juplo.kafka.wordcount.splitter.ApplicationTests.TOPIC_IN;
+import static de.juplo.kafka.wordcount.splitter.ApplicationTests.TOPIC_OUT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+@SpringBootTest(
+		properties = {
+				"juplo.wordcount.splitter.bootstrap-server=${spring.embedded.kafka.brokers}",
+				"juplo.wordcount.splitter.input-topic=" + TOPIC_IN,
+				"juplo.wordcount.splitter.output-topic=" + TOPIC_OUT,
+				"spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
+		})
+@EmbeddedKafka(
+		topics = { TOPIC_IN, TOPIC_OUT },
+		brokerProperties = {
+				"transaction.state.log.replication.factor=1",
+				"transaction.state.log.min.isr=1",
+		})
 class ApplicationTests
 {
+	final static String TOPIC_IN = "in";
+	final static String TOPIC_OUT = "out";
+
+	@Autowired
+	KafkaTemplate<String, String> kafkaTemplate;
+	@Autowired
+	Listener listener;
+
+	@BeforeEach
+	public void clear()
+	{
+		listener.keys.clear();
+		listener.words.clear();
+	}
+
+
 	@Test
 	void contextLoads()
 	{
 	}
+
+	@Test
+	void split()
+	{
+		kafkaTemplate.send(TOPIC_IN, "beate", "Hello World!");
+
+		Awaitility
+				.await("Receive two words")
+				.atMost(Duration.ofSeconds(5))
+				.untilAsserted(() ->
+						assertThat(listener.words.size())
+								.describedAs("Received two words")
+								.isEqualTo(2));
+
+		assertThat(listener.keys)
+				.describedAs("Received unexpected keys")
+				.containsExactly("beate", "beate");
+		assertThat(listener.words)
+				.describedAs("Received unexpected words")
+				.containsExactly("Hello", "World");
+	}
+
+
+	static class Listener
+	{
+		final List<String> keys = new LinkedList<>();
+		final List<String> words = new LinkedList<>();
+
+		@KafkaListener(groupId = "peter", topics = TOPIC_OUT)
+		public void receive(ConsumerRecord<String, String> record)
+		{
+			keys.add(record.key());
+			words.add(record.value());
+		}
+	}
+
+	@TestConfiguration
+	static class Configuration
+	{
+		@Bean
+		Listener listener()
+		{
+			return new Listener();
+		}
+	}
 }