juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessorTopologyTest.java

   1 package de.juplo.kafka.wordcount.counter;
   2
   3 import de.juplo.kafka.wordcount.splitter.TestInputWord;
   4 import de.juplo.kafka.wordcount.top10.TestOutputWord;
   5 import de.juplo.kafka.wordcount.top10.TestOutputWordCounter;
   6 import lombok.extern.slf4j.Slf4j;
   7 import org.apache.kafka.common.serialization.StringSerializer;
   8 import org.apache.kafka.streams.TestInputTopic;
   9 import org.apache.kafka.streams.TestOutputTopic;
  10 import org.apache.kafka.streams.Topology;
  11 import org.apache.kafka.streams.TopologyTestDriver;
  12 import org.apache.kafka.streams.state.Stores;
  13 import org.junit.jupiter.api.AfterEach;
  14 import org.junit.jupiter.api.BeforeEach;
  15 import org.junit.jupiter.api.Test;
  16 import org.springframework.kafka.support.serializer.JsonDeserializer;
  17 import org.springframework.kafka.support.serializer.JsonSerializer;
  18 import org.springframework.util.LinkedMultiValueMap;
  19 import org.springframework.util.MultiValueMap;
  20
  21 import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
  22
  23
  24 @Slf4j
  25 public class CounterStreamProcessorTopologyTest
  26 {
  27   public final static String IN = "TEST-IN";
  28   public final static String OUT = "TEST-OUT";
  29
  30
  31   TopologyTestDriver testDriver;
  32   TestInputTopic<String, TestInputWord> in;
  33   TestOutputTopic<TestOutputWord, TestOutputWordCounter> out;
  34
  35
  36   @BeforeEach
  37   public void setUpTestDriver()
  38   {
  39     Topology topology = CounterStreamProcessor.buildTopology(
  40         IN,
  41         OUT,
  42         Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
  43
  44     testDriver = new TopologyTestDriver(topology, serializationConfig());
  45
  46     in = testDriver.createInputTopic(
  47         IN,
  48         new StringSerializer(),
  49         new JsonSerializer().noTypeInfo());
  50
  51     out = testDriver.createOutputTopic(
  52         OUT,
  53         new JsonDeserializer()
  54             .copyWithType(TestOutputWord.class)
  55             .ignoreTypeHeaders(),
  56         new JsonDeserializer()
  57             .copyWithType(TestOutputWordCounter.class)
  58             .ignoreTypeHeaders());
  59   }
  60
  61
  62   @Test
  63   public void test()
  64   {
  65     TestData
  66         .getInputMessages()
  67         .forEach(word -> in.pipeInput(word.getUser(), word));
  68
  69     MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
  70     out
  71         .readRecordsToList()
  72         .forEach(record -> receivedMessages.add(record.key(), record.value()));
  73
  74     TestData.assertExpectedMessages(receivedMessages);
  75   }
  76
  77   @AfterEach
  78   public void tearDown()
  79   {
  80     testDriver.close();
  81   }
  82 }