juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessorTopologyTest.java

   1 package de.juplo.kafka.wordcount.top10;
   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.KeyValueStore;
   9 import org.apache.kafka.streams.state.Stores;
  10 import org.junit.jupiter.api.AfterEach;
  11 import org.junit.jupiter.api.BeforeEach;
  12 import org.junit.jupiter.api.Test;
  13 import org.springframework.kafka.support.serializer.JsonDeserializer;
  14 import org.springframework.kafka.support.serializer.JsonSerde;
  15 import org.springframework.kafka.support.serializer.JsonSerializer;
  16 import org.springframework.util.LinkedMultiValueMap;
  17 import org.springframework.util.MultiValueMap;
  18
  19 import java.util.Map;
  20 import java.util.Properties;
  21 import java.util.stream.Stream;
  22
  23 import static de.juplo.kafka.wordcount.top10.Top10ApplicationConfiguration.serializationConfig;
  24
  25
  26 @Slf4j
  27 public class Top10StreamProcessorTopologyTest
  28 {
  29   public static final String IN = "TEST-IN";
  30   public static final String OUT = "TEST-OUT";
  31   public static final String STORE_NAME = "TOPOLOGY-TEST";
  32
  33
  34   TopologyTestDriver testDriver;
  35   TestInputTopic<Key, Entry> in;
  36   TestOutputTopic<User, Ranking> out;
  37
  38
  39   @BeforeEach
  40   public void setUp()
  41   {
  42     Topology topology = Top10StreamProcessor.buildTopology(
  43         IN,
  44         OUT,
  45         Stores.inMemoryKeyValueStore(STORE_NAME));
  46
  47     Map<String, Object> propertyMap = serializationConfig();
  48
  49     Properties properties = new Properties();
  50     properties.putAll(propertyMap);
  51
  52     JsonSerde<?> keySerde = new JsonSerde<>();
  53     keySerde.configure(propertyMap, true);
  54     JsonSerde<?> valueSerde = new JsonSerde<>();
  55     valueSerde.configure(propertyMap, false);
  56
  57     testDriver = new TopologyTestDriver(topology, properties);
  58
  59     in = testDriver.createInputTopic(
  60         IN,
  61         (JsonSerializer<Key>)keySerde.serializer(),
  62         (JsonSerializer<Entry>)valueSerde.serializer());
  63
  64     out = testDriver.createOutputTopic(
  65         OUT,
  66         (JsonDeserializer<User>)keySerde.deserializer(),
  67         (JsonDeserializer<Ranking>)valueSerde.deserializer());
  68
  69   }
  70
  71
  72   @Test
  73   public void test()
  74   {
  75     Stream
  76         .of(TestData.INPUT_MESSAGES)
  77         .forEach(kv -> in.pipeInput(
  78             Key.of(kv.key.getUser(), kv.key.getWord()),
  79             Entry.of(kv.value.getWord(), kv.value.getCounter())));
  80
  81     MultiValueMap<User, Ranking> receivedMessages = new LinkedMultiValueMap<>();
  82     out
  83         .readRecordsToList()
  84         .forEach(record -> receivedMessages.add(record.key(), record.value()));
  85
  86     TestData.assertExpectedMessages(receivedMessages);
  87
  88     TestData.assertExpectedNumberOfMessagesForUsers(receivedMessages);
  89     TestData.assertExpectedLastMessagesForUsers(receivedMessages);
  90
  91     KeyValueStore<User, Ranking> store = testDriver.getKeyValueStore(STORE_NAME);
  92     TestData.assertExpectedState(store);
  93   }
  94
  95   @AfterEach
  96   public void tearDown()
  97   {
  98     testDriver.close();
  99   }
 100 }