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.TestData.convertToMap;
  24 import static de.juplo.kafka.wordcount.top10.TestData.parseHeader;
  25 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME;
  26 import static org.springframework.kafka.support.mapping.AbstractJavaTypeMapper.KEY_DEFAULT_CLASSID_FIELD_NAME;
  27
  28
  29 @Slf4j
  30 public class Top10StreamProcessorTopologyTest
  31 {
  32   public static final String IN = "TEST-IN";
  33   public static final String OUT = "TEST-OUT";
  34   public static final String STORE_NAME = "TOPOLOGY-TEST";
  35
  36
  37   TopologyTestDriver testDriver;
  38   TestInputTopic<Key, Entry> in;
  39   TestOutputTopic<User, Ranking> out;
  40
  41
  42   @BeforeEach
  43   public void setUp()
  44   {
  45     Topology topology = Top10StreamProcessor.buildTopology(
  46         IN,
  47         OUT,
  48         Stores.inMemoryKeyValueStore(STORE_NAME));
  49
  50     Top10ApplicationConfiguration applicationConfiguriation =
  51         new Top10ApplicationConfiguration();
  52     Properties streamProcessorProperties =
  53         applicationConfiguriation.streamProcessorProperties(new Top10ApplicationProperties());
  54     Map<String, Object> propertyMap = convertToMap(streamProcessorProperties);
  55
  56     JsonSerde<?> keySerde = new JsonSerde<>();
  57     keySerde.configure(propertyMap, true);
  58     JsonSerde<?> valueSerde = new JsonSerde<>();
  59     valueSerde.configure(propertyMap, false);
  60
  61     testDriver = new TopologyTestDriver(topology, streamProcessorProperties);
  62
  63     in = testDriver.createInputTopic(
  64         IN,
  65         (JsonSerializer<Key>)keySerde.serializer(),
  66         (JsonSerializer<Entry>)valueSerde.serializer());
  67
  68     out = testDriver.createOutputTopic(
  69         OUT,
  70         (JsonDeserializer<User>)keySerde.deserializer(),
  71         (JsonDeserializer<Ranking>)valueSerde.deserializer());
  72
  73   }
  74
  75
  76   @Test
  77   public void test()
  78   {
  79     Stream
  80         .of(TestData.INPUT_MESSAGES)
  81         .forEach(kv -> in.pipeInput(
  82             Key.of(kv.key.getUser(), kv.key.getWord()),
  83             Entry.of(kv.value.getWord(), kv.value.getCounter())));
  84
  85     MultiValueMap<User, Ranking> receivedMessages = new LinkedMultiValueMap<>();
  86     out
  87         .readRecordsToList()
  88         .forEach(record ->
  89         {
  90           log.debug(
  91               "OUT: {} -> {}, {}, {}",
  92               record.key(),
  93               record.value(),
  94               parseHeader(record.headers(), KEY_DEFAULT_CLASSID_FIELD_NAME),
  95               parseHeader(record.headers(), DEFAULT_CLASSID_FIELD_NAME));
  96           receivedMessages.add(record.key(), record.value());
  97         });
  98
  99     TestData.assertExpectedMessages(receivedMessages);
 100
 101     TestData.assertExpectedLastMessagesForUsers(receivedMessages);
 102
 103     KeyValueStore<User, Ranking> store = testDriver.getKeyValueStore(STORE_NAME);
 104     TestData.assertExpectedState(store);
 105   }
 106
 107   @AfterEach
 108   public void tearDown()
 109   {
 110     testDriver.close();
 111   }
 112 }