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 import static de.juplo.kafka.wordcount.counter.TestData.parseHeader;
  23 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
  24 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
  25
  26
  27 @Slf4j
  28 public class CounterStreamProcessorTopologyTest
  29 {
  30   public final static String IN = "TEST-IN";
  31   public final static String OUT = "TEST-OUT";
  32
  33
  34   TopologyTestDriver testDriver;
  35   TestInputTopic<String, TestInputWord> in;
  36   TestOutputTopic<TestOutputWord, TestOutputWordCounter> out;
  37
  38
  39   @BeforeEach
  40   public void setUpTestDriver()
  41   {
  42     Topology topology = CounterStreamProcessor.buildTopology(
  43         IN,
  44         OUT,
  45         Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
  46
  47     testDriver = new TopologyTestDriver(topology, serializationConfig());
  48
  49     in = testDriver.createInputTopic(
  50         IN,
  51         new StringSerializer(),
  52         new JsonSerializer().noTypeInfo());
  53
  54     out = testDriver.createOutputTopic(
  55         OUT,
  56         new JsonDeserializer()
  57             .copyWithType(TestOutputWord.class)
  58             .ignoreTypeHeaders(),
  59         new JsonDeserializer()
  60             .copyWithType(TestOutputWordCounter.class)
  61             .ignoreTypeHeaders());
  62   }
  63
  64
  65   @Test
  66   public void test()
  67   {
  68     TestData
  69         .getInputMessages()
  70         .forEach(word -> in.pipeInput(word.getUser(), word));
  71
  72     MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
  73     out
  74         .readRecordsToList()
  75         .forEach(record ->
  76         {
  77           log.debug(
  78               "OUT: {} -> {}, {}, {}",
  79               record.key(),
  80               record.value(),
  81               parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
  82               parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
  83           receivedMessages.add(record.key(), record.value());
  84         });
  85
  86     TestData.assertExpectedMessages(receivedMessages);
  87   }
  88
  89   @AfterEach
  90   public void tearDown()
  91   {
  92     testDriver.close();
  93   }
  94 }