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.TestInputUser;
   4 import de.juplo.kafka.wordcount.splitter.TestInputWord;
   5 import de.juplo.kafka.wordcount.top10.TestOutputWord;
   6 import de.juplo.kafka.wordcount.top10.TestOutputWordCounter;
   7 import lombok.extern.slf4j.Slf4j;
   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.KeyValueStore;
  13 import org.apache.kafka.streams.state.Stores;
  14 import org.junit.jupiter.api.AfterEach;
  15 import org.junit.jupiter.api.BeforeEach;
  16 import org.junit.jupiter.api.Test;
  17 import org.springframework.kafka.support.serializer.JsonDeserializer;
  18 import org.springframework.kafka.support.serializer.JsonSerializer;
  19 import org.springframework.util.LinkedMultiValueMap;
  20 import org.springframework.util.MultiValueMap;
  21
  22 import java.util.Map;
  23
  24 import static de.juplo.kafka.wordcount.counter.CounterApplicationConfiguriation.serializationConfig;
  25 import static de.juplo.kafka.wordcount.counter.CounterStreamProcessor.STORE_NAME;
  26
  27
  28 @Slf4j
  29 public class CounterStreamProcessorTopologyTest
  30 {
  31   public static final String IN = "TEST-IN";
  32   public static final String OUT = "TEST-OUT";
  33
  34
  35   TopologyTestDriver testDriver;
  36   TestInputTopic<TestInputUser, TestInputWord> in;
  37   TestOutputTopic<TestOutputWord, TestOutputWordCounter> out;
  38
  39
  40   @BeforeEach
  41   public void setUpTestDriver()
  42   {
  43     Topology topology = CounterStreamProcessor.buildTopology(
  44         IN,
  45         OUT,
  46         Stores.inMemoryKeyValueStore(STORE_NAME));
  47
  48     testDriver = new TopologyTestDriver(topology, serializationConfig());
  49
  50     in = testDriver.createInputTopic(IN, serializer(), serializer());
  51     out = testDriver.createOutputTopic(OUT, keyDeserializer(), valueDeserializer());
  52   }
  53
  54
  55   @Test
  56   public void test()
  57   {
  58     TestData
  59         .getInputMessages()
  60         .forEach(kv -> in.pipeInput(kv.key, kv.value));
  61
  62     MultiValueMap<TestOutputWord, TestOutputWordCounter> receivedMessages = new LinkedMultiValueMap<>();
  63     out
  64         .readRecordsToList()
  65         .forEach(record -> receivedMessages.add(record.key(), record.value()));
  66
  67     TestData.assertExpectedMessages(receivedMessages);
  68
  69     TestData.assertExpectedNumberOfMessagesForWord(receivedMessages);
  70     TestData.assertExpectedLastMessagesForWord(receivedMessages);
  71
  72     KeyValueStore<Word, Long> store = testDriver.getKeyValueStore(STORE_NAME);
  73     TestData.assertExpectedState(store);
  74   }
  75
  76   @AfterEach
  77   public void tearDown()
  78   {
  79     testDriver.close();
  80   }
  81
  82
  83   private static JsonSerializer serializer()
  84   {
  85     return new JsonSerializer().noTypeInfo();
  86   }
  87
  88   private JsonDeserializer<TestOutputWord> keyDeserializer()
  89   {
  90     return deserializer(true);
  91   }
  92
  93   private static JsonDeserializer<TestOutputWordCounter> valueDeserializer()
  94   {
  95     return deserializer(false);
  96   }
  97
  98   private static <T> JsonDeserializer<T> deserializer(boolean isKey)
  99   {
 100     JsonDeserializer<T> deserializer = new JsonDeserializer<>();
 101     deserializer.configure(
 102         Map.of(JsonDeserializer.TYPE_MAPPINGS, typeMappingsConfig()),
 103         isKey);
 104     return deserializer;
 105   }
 106
 107   private static String typeMappingsConfig()
 108   {
 109     return CounterStreamProcessor.typeMappingsConfig(TestOutputWord.class, TestOutputWordCounter.class);
 110   }
 111 }