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

   1 package de.juplo.kafka.chat.backend;
   2
   3 import de.juplo.kafka.chat.backend.persistence.kafka.ChatRoomChannel;
   4 import de.juplo.kafka.chat.backend.persistence.kafka.KafkaServicesApplicationRunner;
   5 import lombok.extern.slf4j.Slf4j;
   6 import org.apache.kafka.clients.consumer.Consumer;
   7 import org.apache.kafka.clients.producer.ProducerRecord;
   8 import org.apache.kafka.common.TopicPartition;
   9 import org.junit.jupiter.api.AfterAll;
  10 import org.junit.jupiter.api.BeforeAll;
  11 import org.springframework.beans.factory.annotation.Autowired;
  12 import org.springframework.boot.test.context.SpringBootTest;
  13 import org.springframework.boot.test.mock.mockito.MockBean;
  14 import org.springframework.kafka.core.KafkaTemplate;
  15 import org.springframework.kafka.support.SendResult;
  16 import org.springframework.kafka.test.context.EmbeddedKafka;
  17 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  18
  19 import java.util.List;
  20 import java.util.concurrent.CompletableFuture;
  21
  22 import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.TOPIC;
  23
  24
  25 @SpringBootTest(
  26     webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
  27     properties = {
  28         "chat.backend.services=kafka",
  29         "chat.backend.kafka.client-id-PREFIX=TEST",
  30         "chat.backend.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  31         "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
  32         "chat.backend.kafka.chatroom-channel-topic=" + TOPIC,
  33         "chat.backend.kafka.num-partitions=10",
  34         })
  35 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
  36 @Slf4j
  37 class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
  38 {
  39   final static String TOPIC = "TEST_CHATROOM_CHANNEL";
  40
  41   static CompletableFuture<Void> CONSUMER_JOB;
  42
  43   @MockBean
  44   KafkaServicesApplicationRunner kafkaServicesApplicationRunner;
  45
  46   @BeforeAll
  47   public static void sendAndLoadStoredData(
  48       @Autowired KafkaTemplate<String, String> messageTemplate,
  49       @Autowired Consumer chatRoomChannelConsumer,
  50       @Autowired ThreadPoolTaskExecutor taskExecutor,
  51       @Autowired ChatRoomChannel chatRoomChannel)
  52   {
  53     send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
  54     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
  55     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
  56     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
  57     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
  58
  59     List<TopicPartition> assignedPartitions = List.of(new TopicPartition(TOPIC, 2));
  60     chatRoomChannelConsumer.assign(assignedPartitions);
  61     chatRoomChannel.onPartitionsAssigned(assignedPartitions);
  62     CONSUMER_JOB = taskExecutor
  63         .submitCompletable(chatRoomChannel)
  64         .exceptionally(e ->
  65         {
  66           log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
  67           return null;
  68         });
  69   }
  70
  71   static void send(KafkaTemplate<String, String> kafkaTemplate, String key, String value, String typeId)
  72   {
  73     ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
  74     record.headers().add("__TypeId__", typeId.getBytes());
  75     SendResult<String, String> result = kafkaTemplate.send(record).join();
  76     log.info(
  77         "Sent {}={} to {}",
  78         key,
  79         value,
  80         new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
  81   }
  82
  83   @AfterAll
  84   static void joinConsumerJob(@Autowired Consumer chatRoomChannelConsumer)
  85   {
  86     log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
  87     chatRoomChannelConsumer.wakeup();
  88     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
  89     CONSUMER_JOB.join();
  90     log.info("Joined the consumer of the ChatRoomChannel");
  91   }
  92 }