juplo.de Git - demos/kafka/chat/blob

   1 package de.juplo.kafka.chat.backend.implementation.kafka;
   2
   3 import de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest;
   4 import lombok.extern.slf4j.Slf4j;
   5 import org.apache.kafka.common.TopicPartition;
   6 import org.junit.jupiter.api.BeforeAll;
   7 import org.springframework.beans.factory.annotation.Autowired;
   8 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
   9 import org.springframework.boot.autoconfigure.task.TaskExecutionAutoConfiguration;
  10 import org.springframework.boot.test.context.TestConfiguration;
  11 import org.springframework.context.annotation.Bean;
  12 import org.springframework.kafka.core.KafkaTemplate;
  13 import org.springframework.kafka.test.context.EmbeddedKafka;
  14 import org.springframework.test.context.ContextConfiguration;
  15 import org.springframework.test.context.TestPropertySource;
  16
  17 import java.util.List;
  18
  19 import static de.juplo.kafka.chat.backend.domain.ChatHomeServiceWithShardsTest.NUM_SHARDS;
  20 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.DATA_TOPIC;
  21 import static de.juplo.kafka.chat.backend.implementation.kafka.KafkaChatHomeServiceTest.INFO_TOPIC;
  22
  23
  24 @ContextConfiguration(classes = {
  25         KafkaTestUtils.KafkaTestConfiguration.class,
  26         KafkaAutoConfiguration.class,
  27         TaskExecutionAutoConfiguration.class,
  28     })
  29 @TestPropertySource(properties = {
  30         "chat.backend.services=kafka",
  31         "chat.backend.kafka.client-id-PREFIX=TEST",
  32         "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  33         "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  34         "chat.backend.kafka.info-channel-topic=" + INFO_TOPIC,
  35         "chat.backend.kafka.data-channel-topic=" + DATA_TOPIC,
  36         "chat.backend.kafka.num-partitions=" + NUM_SHARDS,
  37 })
  38 @EmbeddedKafka(
  39     topics = { INFO_TOPIC, DATA_TOPIC },
  40     partitions = NUM_SHARDS)
  41 @Slf4j
  42 public class KafkaChatHomeServiceTest extends ChatHomeServiceWithShardsTest
  43 {
  44   final static String INFO_TOPIC = "KAFKA_CHAT_HOME_TEST_INFO";
  45   final static String DATA_TOPIC = "KAFKA_CHAT_HOME_TEST_DATA";
  46
  47
  48   @BeforeAll
  49   static void sendAndLoadStoredData(
  50       @Autowired KafkaTemplate<String, String> messageTemplate,
  51       @Autowired ChannelTaskExecutor infoChannelTaskExecutor,
  52       @Autowired ChannelTaskExecutor dataChannelTaskExecutor)
  53   {
  54     KafkaTestUtils.initKafkaSetup(
  55         INFO_TOPIC,
  56         DATA_TOPIC,
  57         messageTemplate,
  58         infoChannelTaskExecutor,
  59         dataChannelTaskExecutor);
  60   }
  61
  62
  63   @TestConfiguration
  64   static class KafkaChatHomeServiceTestConfiguration
  65   {
  66     @Bean
  67     WorkAssignor infoChannelWorkAssignor()
  68     {
  69       return consumer ->
  70       {
  71         List<TopicPartition> partitions = consumer
  72             .partitionsFor(INFO_TOPIC)
  73             .stream()
  74             .map(partitionInfo -> new TopicPartition(INFO_TOPIC, partitionInfo.partition()))
  75             .toList();
  76         consumer.assign(partitions);
  77       };
  78     }
  79   }
  80 }