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.common.serialization.StringDeserializer;
   4 import org.apache.kafka.common.serialization.StringSerializer;
   5 import org.apache.kafka.streams.TestInputTopic;
   6 import org.apache.kafka.streams.TestOutputTopic;
   7 import org.apache.kafka.streams.Topology;
   8 import org.apache.kafka.streams.TopologyTestDriver;
   9 import org.apache.kafka.streams.state.Stores;
  10 import org.junit.jupiter.api.Test;
  11
  12 import java.util.List;
  13 import java.util.Properties;
  14
  15
  16 public class CounterStreamProcessorTopologyTest
  17 {
  18   public final static String IN = "TEST-IN";
  19   public final static String OUT = "TEST-OUT";
  20
  21   @Test
  22   public void test()
  23   {
  24     Topology topology = CounterStreamProcessor.buildTopology(
  25         IN,
  26         OUT,
  27         Stores.inMemoryKeyValueStore("TOPOLOGY-TEST"));
  28
  29     CounterApplicationConfiguriation config =
  30         new CounterApplicationConfiguriation();
  31     Properties properties =
  32         config.propertyMap(new CounterApplicationProperties());
  33
  34     TopologyTestDriver testDriver = new TopologyTestDriver(topology, properties);
  35
  36     TestInputTopic<String, String> in = testDriver.createInputTopic(
  37         IN,
  38         new StringSerializer(),
  39         new StringSerializer());
  40
  41     TestOutputTopic<String, String> out = testDriver.createOutputTopic(
  42         OUT,
  43         new StringDeserializer(),
  44         new StringDeserializer());
  45
  46     TestData.writeInputData((key, value) -> in.pipeInput(key, value));
  47
  48     List<Message> receivedMessages = out
  49         .readRecordsToList()
  50         .stream()
  51         .map(record -> Message.of(record.key(), record.value()))
  52         .toList();
  53
  54     TestData.assertExpectedResult(receivedMessages);
  55   }
  56 }