juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/splitter/ApplicationTests.java

   1 package de.juplo.kafka.wordcount.splitter;
   2
   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;
  14
  15 import java.time.Duration;
  16 import java.util.LinkedList;
  17 import java.util.List;
  18
  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;
  22
  23
  24 @SpringBootTest(
  25                 properties = {
  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}",
  30                 })
  31 @EmbeddedKafka(
  32                 topics = { TOPIC_IN, TOPIC_OUT },
  33                 brokerProperties = {
  34                                 "transaction.state.log.replication.factor=1",
  35                                 "transaction.state.log.min.isr=1",
  36                 })
  37 class ApplicationTests
  38 {
  39         final static String TOPIC_IN = "in";
  40         final static String TOPIC_OUT = "out";
  41
  42         @Autowired
  43         KafkaTemplate<String, String> kafkaTemplate;
  44         @Autowired
  45         Listener listener;
  46
  47         @BeforeEach
  48         public void clear()
  49         {
  50                 listener.keys.clear();
  51                 listener.words.clear();
  52         }
  53
  54
  55         @Test
  56         void contextLoads()
  57         {
  58         }
  59
  60         @Test
  61         void split()
  62         {
  63                 kafkaTemplate.send(TOPIC_IN, "beate", "Hello World!");
  64
  65                 Awaitility
  66                                 .await("Receive two words")
  67                                 .atMost(Duration.ofSeconds(5))
  68                                 .untilAsserted(() ->
  69                                                 assertThat(listener.words.size())
  70                                                                 .describedAs("Received two words")
  71                                                                 .isEqualTo(2));
  72
  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");
  79         }
  80
  81
  82         static class Listener
  83         {
  84                 final List<String> keys = new LinkedList<>();
  85                 final List<String> words = new LinkedList<>();
  86
  87                 @KafkaListener(groupId = "peter", topics = TOPIC_OUT)
  88                 public void receive(ConsumerRecord<String, String> record)
  89                 {
  90                         keys.add(record.key());
  91                         words.add(record.value());
  92                 }
  93         }
  94
  95         @TestConfiguration
  96         static class Configuration
  97         {
  98                 @Bean
  99                 Listener listener()
 100                 {
 101                         return new Listener();
 102                 }
 103         }
 104 }