1 package de.juplo.kafka.wordcount.splitter;
3 import org.apache.kafka.clients.consumer.ConsumerRecord;
4 import org.awaitility.Awaitility;
5 import org.junit.jupiter.api.BeforeEach;
6 import org.junit.jupiter.api.Test;
7 import org.springframework.beans.factory.annotation.Autowired;
8 import org.springframework.boot.test.context.SpringBootTest;
9 import org.springframework.boot.test.context.TestConfiguration;
10 import org.springframework.context.annotation.Bean;
11 import org.springframework.kafka.annotation.KafkaListener;
12 import org.springframework.kafka.core.KafkaTemplate;
13 import org.springframework.kafka.test.context.EmbeddedKafka;
15 import java.time.Duration;
16 import java.util.LinkedList;
17 import java.util.List;
19 import static de.juplo.kafka.wordcount.splitter.ApplicationTests.TOPIC_IN;
20 import static de.juplo.kafka.wordcount.splitter.ApplicationTests.TOPIC_OUT;
21 import static org.assertj.core.api.Assertions.assertThat;
26 "juplo.wordcount.splitter.bootstrap-server=${spring.embedded.kafka.brokers}",
27 "juplo.wordcount.splitter.input-topic=" + TOPIC_IN,
28 "juplo.wordcount.splitter.output-topic=" + TOPIC_OUT,
29 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
32 topics = { TOPIC_IN, TOPIC_OUT },
34 "transaction.state.log.replication.factor=1",
35 "transaction.state.log.min.isr=1",
37 class ApplicationTests
39 final static String TOPIC_IN = "in";
40 final static String TOPIC_OUT = "out";
43 KafkaTemplate<String, String> kafkaTemplate;
50 listener.keys.clear();
51 listener.words.clear();
63 kafkaTemplate.send(TOPIC_IN, "beate", "Hello World!");
66 .await("Receive two words")
67 .atMost(Duration.ofSeconds(5))
69 assertThat(listener.words.size())
70 .describedAs("Received two words")
73 assertThat(listener.keys)
74 .describedAs("Received unexpected keys")
75 .containsExactly("beate", "beate");
76 assertThat(listener.words)
77 .describedAs("Received unexpected words")
78 .containsExactly("Hello", "World");
84 final List<String> keys = new LinkedList<>();
85 final List<String> words = new LinkedList<>();
87 @KafkaListener(groupId = "peter", topics = TOPIC_OUT)
88 public void receive(ConsumerRecord<String, String> record)
90 keys.add(record.key());
91 words.add(record.value());
96 static class Configuration
101 return new Listener();