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 org.apache.kafka.streams.TestInputTopic;
   4 import org.apache.kafka.streams.TestOutputTopic;
   5 import org.apache.kafka.streams.Topology;
   6 import org.apache.kafka.streams.TopologyTestDriver;
   7 import org.apache.kafka.streams.state.Stores;
   8 import org.junit.jupiter.api.Test;
   9 import org.springframework.kafka.support.serializer.JsonDeserializer;
  10 import org.springframework.kafka.support.serializer.JsonSerde;
  11 import org.springframework.kafka.support.serializer.JsonSerializer;
  12
  13 import java.util.List;
  14 import java.util.Map;
  15 import java.util.Properties;
  16 import java.util.stream.Collectors;
  17
  18
  19 public class CounterStreamProcessorTopologyTest
  20 {
  21   public final static String IN = "TEST-IN";
  22   public final static String OUT = "TEST-OUT";
  23
  24   @Test
  25   public void test()
  26   {
  27     Topology topology = CounterStreamProcessor.buildTopology(
  28         IN,
  29         OUT,
  30         Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
  31
  32     CounterApplicationConfiguriation applicationConfiguriation =
  33         new CounterApplicationConfiguriation();
  34     Properties streamProcessorProperties =
  35         applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
  36
  37     Map<String, ?> propertyMap = streamProcessorProperties
  38         .entrySet()
  39         .stream()
  40         .collect(
  41             Collectors.toMap(
  42                 entry -> (String)entry.getKey(),
  43                 entry -> entry.getValue()
  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, WordCount> out = testDriver.createOutputTopic(
  58         OUT,
  59         (JsonDeserializer<Word>)keySerde.deserializer(),
  60         (JsonDeserializer<WordCount>)valueSerde.deserializer());
  61
  62     TestData.writeInputData((key, value) -> in.pipeInput(key, value));
  63
  64     List<Message> receivedMessages = out
  65         .readRecordsToList()
  66         .stream()
  67         .map(record -> Message.of(record.key(), record.value()))
  68         .toList();
  69
  70     TestData.assertExpectedResult(receivedMessages);
  71   }
  72 }