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 lombok.extern.slf4j.Slf4j;
   4 import org.apache.kafka.streams.TestInputTopic;
   5 import org.apache.kafka.streams.TestOutputTopic;
   6 import org.apache.kafka.streams.Topology;
   7 import org.apache.kafka.streams.TopologyTestDriver;
   8 import org.apache.kafka.streams.state.Stores;
   9 import org.junit.jupiter.api.Test;
  10 import org.springframework.kafka.support.serializer.JsonDeserializer;
  11 import org.springframework.kafka.support.serializer.JsonSerde;
  12 import org.springframework.kafka.support.serializer.JsonSerializer;
  13 import org.springframework.util.LinkedMultiValueMap;
  14 import org.springframework.util.MultiValueMap;
  15
  16 import java.util.Map;
  17 import java.util.Properties;
  18
  19 import static de.juplo.kafka.wordcount.counter.TestData.convertToMap;
  20 import static de.juplo.kafka.wordcount.counter.TestData.parseHeader;
  21 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
  22 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
  23
  24
  25 @Slf4j
  26 public class CounterStreamProcessorTopologyTest
  27 {
  28   public final static String IN = "TEST-IN";
  29   public final static String OUT = "TEST-OUT";
  30
  31   @Test
  32   public void test()
  33   {
  34     Topology topology = CounterStreamProcessor.buildTopology(
  35         IN,
  36         OUT,
  37         Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
  38
  39     CounterApplicationConfiguriation applicationConfiguriation =
  40         new CounterApplicationConfiguriation();
  41     Properties streamProcessorProperties =
  42         applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
  43     Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
  44
  45     JsonSerde<?> keySerde = new JsonSerde<>();
  46     keySerde.configure(propertyMap, true);
  47     JsonSerde<?> valueSerde = new JsonSerde<>();
  48     valueSerde.configure(propertyMap, false);
  49
  50     TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
  51
  52     TestInputTopic<String, Word> in = testDriver.createInputTopic(
  53         IN,
  54         (JsonSerializer<String>)keySerde.serializer(),
  55         (JsonSerializer<Word>)valueSerde.serializer());
  56
  57     TestOutputTopic<Word, WordCounter> out = testDriver.createOutputTopic(
  58         OUT,
  59         (JsonDeserializer<Word>)keySerde.deserializer(),
  60         (JsonDeserializer<WordCounter>)valueSerde.deserializer());
  61
  62     TestData.injectInputMessages((key, value) -> in.pipeInput(key, value));
  63
  64     MultiValueMap<Word, WordCounter> receivedMessages = new LinkedMultiValueMap<>();
  65     out
  66         .readRecordsToList()
  67         .forEach(record ->
  68         {
  69           log.debug(
  70               "OUT: {} -> {}, {}, {}",
  71               record.key(),
  72               record.value(),
  73               parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
  74               parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
  75           receivedMessages.add(record.key(), record.value());
  76         });
  77
  78     TestData.assertExpectedMessages(receivedMessages);
  79   }
  80 }