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 SplitterStreamProcessor splitter;
45 KafkaTemplate<String, String> kafkaTemplate;
52 listener.keys.clear();
53 listener.words.clear();
66 kafkaTemplate.send(TOPIC_IN, "beate", "Hello World!");
69 .await("Receive two words")
70 .atMost(Duration.ofSeconds(5))
72 assertThat(listener.words.size())
73 .describedAs("Received two words")
76 assertThat(listener.keys)
77 .describedAs("Received unexpected keys")
78 .containsExactly("beate", "beate");
79 assertThat(listener.words)
80 .describedAs("Received unexpected words")
81 .containsExactly("Hello", "World");
87 final List<String> keys = new LinkedList<>();
88 final List<String> words = new LinkedList<>();
90 @KafkaListener(groupId = "peter", topics = TOPIC_OUT)
91 public void receive(ConsumerRecord<String, String> record)
93 keys.add(record.key());
94 words.add(record.value());
99 static class Configuration
104 return new Listener();