juplo.de Git - demos/kafka/wordcount/blob - src/test/java/de/juplo/kafka/wordcount/query/QueryStreamProcessorTopologyTest.java

   1 package de.juplo.kafka.wordcount.query;
   2
   3 import de.juplo.kafka.wordcount.top10.TestRanking;
   4 import de.juplo.kafka.wordcount.top10.TestUser;
   5 import de.juplo.kafka.wordcount.users.TestUserData;
   6 import lombok.extern.slf4j.Slf4j;
   7 import org.apache.kafka.common.serialization.StringSerializer;
   8 import org.apache.kafka.streams.TestInputTopic;
   9 import org.apache.kafka.streams.Topology;
  10 import org.apache.kafka.streams.TopologyTestDriver;
  11 import org.apache.kafka.streams.state.KeyValueStore;
  12 import org.apache.kafka.streams.state.Stores;
  13 import org.junit.jupiter.api.AfterEach;
  14 import org.junit.jupiter.api.BeforeEach;
  15 import org.junit.jupiter.api.Test;
  16 import org.springframework.kafka.support.serializer.JsonSerializer;
  17
  18 import java.util.Map;
  19
  20 import static de.juplo.kafka.wordcount.query.QueryApplicationConfiguration.serializationConfig;
  21
  22
  23 @Slf4j
  24 public class QueryStreamProcessorTopologyTest
  25 {
  26   public static final String TOP10_IN = "TOP10-IN";
  27   public static final String USERS_IN = "USERS-IN";
  28   public static final String RANKING_STORE_NAME = "TOPOLOGY-TEST-RANKINGS";
  29   public static final String USERS_STORE_NAME = "TOPOLOGY-TEST-USERS";
  30
  31
  32   TopologyTestDriver testDriver;
  33   TestInputTopic<TestUser, TestRanking> top10In;
  34   TestInputTopic<String, TestUserData> userIn;
  35
  36
  37   @BeforeEach
  38   public void setUp()
  39   {
  40     Topology topology = QueryStreamProcessor.buildTopology(
  41         USERS_IN,
  42         TOP10_IN,
  43         Stores.inMemoryKeyValueStore(USERS_STORE_NAME),
  44         Stores.inMemoryKeyValueStore(RANKING_STORE_NAME));
  45
  46     testDriver = new TopologyTestDriver(topology, serializationConfig());
  47
  48     top10In = testDriver.createInputTopic(
  49         TOP10_IN,
  50         jsonSerializer(TestUser.class, true),
  51         jsonSerializer(TestRanking.class,false));
  52
  53     userIn = testDriver.createInputTopic(
  54         USERS_IN,
  55         new StringSerializer(),
  56         jsonSerializer(TestUserData.class, false).noTypeInfo());
  57   }
  58
  59
  60   @Test
  61   public void test()
  62   {
  63     TestData
  64         .getUsersMessages()
  65         .forEach(kv -> userIn.pipeInput(kv.key, kv.value));
  66     TestData
  67         .getTop10Messages()
  68         .forEach(kv -> top10In.pipeInput(kv.key, kv.value));
  69
  70     KeyValueStore<String, UserRanking> store = testDriver.getKeyValueStore(RANKING_STORE_NAME);
  71     TestData.assertExpectedState(user -> store.get(user));
  72   }
  73
  74   @AfterEach
  75   public void tearDown()
  76   {
  77     testDriver.close();
  78   }
  79
  80   private <T> JsonSerializer<T> jsonSerializer(Class<T> type, boolean isKey)
  81   {
  82     JsonSerializer<T> jsonSerializer = new JsonSerializer<>();
  83     jsonSerializer.configure(
  84         Map.of(
  85             JsonSerializer.TYPE_MAPPINGS,
  86             "user:" + TestUser.class.getName() + "," +
  87             "ranking:" + TestRanking.class.getName()),
  88         isKey);
  89     return jsonSerializer;
  90   }
  91 }