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.HashMap;
  17 import java.util.LinkedList;
  18 import java.util.List;
  19 import java.util.Map;
  20
  21 import static de.juplo.kafka.wordcount.splitter.ApplicationTests.TOPIC_IN;
  22 import static de.juplo.kafka.wordcount.splitter.ApplicationTests.TOPIC_OUT;
  23 import static org.assertj.core.api.Assertions.assertThat;
  24
  25
  26 @SpringBootTest(
  27                 properties = {
  28                                 "juplo.wordcount.splitter.bootstrap-server=${spring.embedded.kafka.brokers}",
  29                                 "juplo.wordcount.splitter.input-topic=" + TOPIC_IN,
  30                                 "juplo.wordcount.splitter.output-topic=" + TOPIC_OUT,
  31                                 "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  32                 })
  33 @EmbeddedKafka(
  34                 topics = { TOPIC_IN, TOPIC_OUT },
  35                 brokerProperties = {
  36                                 "transaction.state.log.replication.factor=1",
  37                                 "transaction.state.log.min.isr=1",
  38                 })
  39 class ApplicationTests
  40 {
  41         final static String TOPIC_IN = "in";
  42         final static String TOPIC_OUT = "out";
  43
  44         @Autowired
  45         KafkaTemplate<String, String> kafkaTemplate;
  46         @Autowired
  47         Listener listener;
  48         @Autowired
  49         MessageSplitter messagSplitter;
  50
  51         @BeforeEach
  52         public void clear()
  53         {
  54                 listener.reset();
  55         }
  56
  57
  58         @Test
  59         void contextLoads()
  60         {
  61         }
  62
  63         @Test
  64         void split()
  65         {
  66                 kafkaTemplate.send(TOPIC_IN, "beate", LORE_IPSUM);
  67
  68                 String[] words = messagSplitter.split(LORE_IPSUM);
  69
  70                 Awaitility
  71                                 .await("Receive all words")
  72                                 .atMost(Duration.ofSeconds(5))
  73                                 .untilAsserted(() ->
  74                                                 assertThat(listener.total)
  75                                                                 .describedAs("Received all words")
  76                                                                 .isEqualTo(words.length));
  77
  78                 assertThat(listener.words.keySet())
  79                                 .describedAs("Received unexpected key(s)")
  80                                 .containsExactlyInAnyOrder("beate");
  81                 assertThat(listener.words.get("beate"))
  82                                 .describedAs("Received unexpected words")
  83                                 .containsExactly(words);
  84         }
  85
  86         @Test
  87         void order()
  88         {
  89                 kafkaTemplate.send(TOPIC_IN, "beate", LORE_IPSUM);
  90                 kafkaTemplate.send(TOPIC_IN, "peter", LORE_IPSUM);
  91                 kafkaTemplate.send(TOPIC_IN, "klaus", LORE_IPSUM);
  92
  93                 String[] words = messagSplitter.split(LORE_IPSUM);
  94
  95                 Awaitility
  96                                 .await("Receive all words")
  97                                 .atMost(Duration.ofSeconds(5))
  98                                 .untilAsserted(() -> {
  99                                         assertThat(listener.total)
 100                                                         .describedAs("Received all words for beate")
 101                                                         .isEqualTo(words.length * 3);
 102                                 });
 103
 104                 assertThat(listener.words.keySet())
 105                                 .describedAs("Received unexpected key(s)")
 106                                 .containsExactlyInAnyOrder("beate", "peter", "klaus");
 107                 assertThat(listener.words.get("beate"))
 108                                 .describedAs("Received unexpected words for beate")
 109                                 .containsExactly(words);
 110                 assertThat(listener.words.get("peter"))
 111                                 .describedAs("Received unexpected words for beate")
 112                                 .containsExactly(words);
 113                 assertThat(listener.words.get("klaus"))
 114                                 .describedAs("Received unexpected words for beate")
 115                                 .containsExactly(words);
 116         }
 117
 118
 119         static class Listener
 120         {
 121                 final Map<String, List<String>> words = new HashMap<>();
 122                 int total = 0;
 123
 124                 @KafkaListener(groupId = "peter", topics = TOPIC_OUT)
 125                 public void receive(ConsumerRecord<String, String> record)
 126                 {
 127                         List<String> words = this.words.get(record.key());
 128                         if (words == null)
 129                         {
 130                                 words = new LinkedList<>();
 131                                 this.words.put(record.key(), words);
 132                         }
 133                         words.add(record.value());
 134                         total++;
 135                 }
 136
 137                 void reset()
 138                 {
 139                         words.clear();
 140                         total = 0;
 141                 }
 142         }
 143
 144         @TestConfiguration
 145         static class Configuration
 146         {
 147                 @Bean
 148                 Listener listener()
 149                 {
 150                         return new Listener();
 151                 }
 152         }
 153
 154
 155         final static String LORE_IPSUM =
 156                         "Lorem ipsum dolor sit amet, consectetur adipisici elit, sed eiusmod " +
 157                         "tempor incidunt ut labore et dolore magna aliqua. Ut enim ad minim " +
 158                         "veniam, quis nostrud exercitation ullamco laboris nisi ut aliquid ex " +
 159                         "ea commodi consequat. Quis aute iure reprehenderit in voluptate velit " +
 160                         "esse cillum dolore eu fugiat nulla pariatur. Excepteur sint obcaecat " +
 161                         "cupiditat non proident, sunt in culpa qui officia deserunt mollit anim " +
 162                         "id est laborum.";
 163 }