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.*;
   4 import org.apache.kafka.streams.state.Stores;
   5 import org.junit.jupiter.api.Test;
   6 import org.springframework.kafka.support.serializer.JsonDeserializer;
   7 import org.springframework.kafka.support.serializer.JsonSerde;
   8 import org.springframework.kafka.support.serializer.JsonSerializer;
   9
  10 import java.util.List;
  11 import java.util.Map;
  12 import java.util.Properties;
  13
  14 import static de.juplo.kafka.wordcount.counter.TestData.convertToMap;
  15
  16
  17 public class CounterStreamProcessorTopologyTest
  18 {
  19   public final static String IN = "TEST-IN";
  20   public final static String OUT = "TEST-OUT";
  21
  22   @Test
  23   public void test()
  24   {
  25     Topology topology = CounterStreamProcessor.buildTopology(
  26         IN,
  27         OUT,
  28         Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
  29
  30     CounterApplicationConfiguriation applicationConfiguriation =
  31         new CounterApplicationConfiguriation();
  32     Properties streamProcessorProperties =
  33         applicationConfiguriation.streamProcessorProperties(new CounterApplicationProperties());
  34     Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
  35
  36     JsonSerde<?> keySerde = new JsonSerde<>();
  37     keySerde.configure(propertyMap, true);
  38     JsonSerde<?> valueSerde = new JsonSerde<>();
  39     valueSerde.configure(propertyMap, false);
  40
  41     TopologyTestDriver testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
  42
  43     TestInputTopic<String, Word> in = testDriver.createInputTopic(
  44         IN,
  45         (JsonSerializer<String>)keySerde.serializer(),
  46         (JsonSerializer<Word>)valueSerde.serializer());
  47
  48     TestOutputTopic<Word, WordCounter> out = testDriver.createOutputTopic(
  49         OUT,
  50         (JsonDeserializer<Word>)keySerde.deserializer(),
  51         (JsonDeserializer<WordCounter>)valueSerde.deserializer());
  52
  53     TestData.writeInputData((key, value) -> in.pipeInput(key, value));
  54
  55     List<KeyValue<Word, WordCounter>> receivedMessages = out
  56         .readRecordsToList()
  57         .stream()
  58         .map(record -> KeyValue.pair(record.key(), record.value()))
  59         .toList();
  60
  61     TestData.assertExpectedResult(receivedMessages);
  62   }
  63 }