juplo.de Git - demos/kafka/chat/blob - src/test/java/de/juplo/kafka/chat/backend/implementation/kafka/KafkaTestUtils.java

   1 package de.juplo.kafka.chat.backend.implementation.kafka;
   2
   3 import de.juplo.kafka.chat.backend.ChatBackendProperties;
   4 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.apache.kafka.clients.producer.ProducerRecord;
   7 import org.apache.kafka.common.TopicPartition;
   8 import org.junit.jupiter.api.AfterAll;
   9 import org.junit.jupiter.api.BeforeAll;
  10 import org.springframework.beans.factory.annotation.Autowired;
  11 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
  12 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
  13 import org.springframework.boot.context.properties.EnableConfigurationProperties;
  14 import org.springframework.boot.test.context.SpringBootTest;
  15 import org.springframework.boot.test.context.TestConfiguration;
  16 import org.springframework.context.annotation.Bean;
  17 import org.springframework.context.annotation.Import;
  18 import org.springframework.kafka.core.KafkaTemplate;
  19 import org.springframework.kafka.support.SendResult;
  20 import org.springframework.kafka.test.context.EmbeddedKafka;
  21
  22 import java.time.Clock;
  23 import java.util.List;
  24
  25 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
  26 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.DATA_TOPIC;
  27 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.INFO_TOPIC;
  28
  29
  30 @SpringBootTest(
  31     classes = {
  32         KafkaChatHomeServiceTest.KafkaChatHomeTestConfiguration.class,
  33         KafkaAutoConfiguration.class,
  34         TaskExecutionAutoConfiguration.class,
  35     },
  36     properties = {
  37     "spring.main.allow-bean-definition-overriding=true",
  38     "chat.backend.services=kafka",
  39     "chat.backend.kafka.client-id-PREFIX=TEST",
  40     "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  41     "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  42     "chat.backend.kafka.info-channel-topic=" + INFO_TOPIC,
  43     "chat.backend.kafka.data-channel-topic=" + DATA_TOPIC,
  44     "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
  45 })
  46 @EmbeddedKafka(
  47     topics = { INFO_TOPIC, DATA_TOPIC },
  48     partitions = 10)
  49 @Slf4j
  50 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
  51 {
  52   final static String INFO_TOPIC = "KAFKA_CHAT_HOME_TEST_INFO";
  53   final static String DATA_TOPIC = "KAFKA_CHAT_HOME_TEST_DATA";
  54
  55
  56   @TestConfiguration
  57   @EnableConfigurationProperties(ChatBackendProperties.class)
  58   @Import(KafkaServicesConfiguration.class)
  59   static class KafkaChatHomeTestConfiguration
  60   {
  61     @Bean
  62     WorkAssignor dataChannelWorkAssignor(DataChannel dataChannel)
  63     {
  64       return consumer ->
  65       {
  66         List<TopicPartition> assignedPartitions =
  67             List.of(new TopicPartition(DATA_TOPIC, 2));
  68         consumer.assign(assignedPartitions);
  69         dataChannel.onPartitionsAssigned(assignedPartitions);
  70       };
  71     }
  72
  73     @Bean
  74     Clock clock()
  75     {
  76       return Clock.systemDefaultZone();
  77     }
  78   }
  79
  80
  81   @BeforeAll
  82   public static void sendAndLoadStoredData(
  83       @Autowired ConsumerTaskRunner consumerTaskRunner,
  84       @Autowired KafkaTemplate<String, String> messageTemplate) throws InterruptedException
  85   {
  86     send(messageTemplate, INFO_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "event_chatroom_created");
  87     send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
  88     send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
  89     send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
  90     send(messageTemplate, DATA_TOPIC, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
  91
  92     consumerTaskRunner.run();
  93   }
  94
  95   static void send(
  96       KafkaTemplate<String, String> kafkaTemplate,
  97       String topic,
  98       String key,
  99       String value,
 100       String typeId)
 101   {
 102     ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
 103     record.headers().add("__TypeId__", typeId.getBytes());
 104     SendResult<String, String> result = kafkaTemplate.send(record).join();
 105     log.info(
 106         "Sent {}={} to {}",
 107         key,
 108         value,
 109         new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
 110   }
 111
 112   @AfterAll
 113   static void joinConsumerTasks(@Autowired ConsumerTaskRunner consumerTaskRunner)
 114   {
 115     consumerTaskRunner.joinConsumerTasks();
 116   }
 117 }