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         SplitterStreamProcessor splitter;
  44         @Autowired
  45         KafkaTemplate<String, String> kafkaTemplate;
  46         @Autowired
  47         Listener listener;
  48
  49         @BeforeEach
  50         public void clear()
  51         {
  52                 listener.keys.clear();
  53                 listener.words.clear();
  54         }
  55
  56
  57         @Test
  58         void contextLoads()
  59         {
  60                 splitter.stop();
  61         }
  62
  63         @Test
  64         void split()
  65         {
  66                 kafkaTemplate.send(TOPIC_IN, "beate", "Hello World!");
  67
  68                 Awaitility
  69                                 .await("Receive two words")
  70                                 .atMost(Duration.ofSeconds(5))
  71                                 .untilAsserted(() ->
  72                                                 assertThat(listener.words.size())
  73                                                                 .describedAs("Received two words")
  74                                                                 .isEqualTo(2));
  75
  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");
  82         }
  83
  84
  85         static class Listener
  86         {
  87                 final List<String> keys = new LinkedList<>();
  88                 final List<String> words = new LinkedList<>();
  89
  90                 @KafkaListener(groupId = "peter", topics = TOPIC_OUT)
  91                 public void receive(ConsumerRecord<String, String> record)
  92                 {
  93                         keys.add(record.key());
  94                         words.add(record.value());
  95                 }
  96         }
  97
  98         @TestConfiguration
  99         static class Configuration
 100         {
 101                 @Bean
 102                 Listener listener()
 103                 {
 104                         return new Listener();
 105                 }
 106         }
 107 }