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

   1 package de.juplo.kafka.chat.backend.persistence.kafka;
   2
   3 import de.juplo.kafka.chat.backend.ChatBackendProperties;
   4 import de.juplo.kafka.chat.backend.domain.ChatHomeWithShardsTestBase;
   5 import de.juplo.kafka.chat.backend.persistence.kafka.messages.AbstractMessageTo;
   6 import lombok.extern.slf4j.Slf4j;
   7 import org.apache.kafka.clients.consumer.Consumer;
   8 import org.apache.kafka.clients.producer.KafkaProducer;
   9 import org.apache.kafka.clients.producer.Producer;
  10 import org.apache.kafka.clients.producer.ProducerConfig;
  11 import org.apache.kafka.clients.producer.ProducerRecord;
  12 import org.apache.kafka.common.TopicPartition;
  13 import org.apache.kafka.common.serialization.StringSerializer;
  14 import org.junit.jupiter.api.AfterAll;
  15 import org.junit.jupiter.api.BeforeAll;
  16 import org.springframework.beans.factory.annotation.Autowired;
  17 import org.springframework.boot.test.context.TestConfiguration;
  18 import org.springframework.context.annotation.Bean;
  19 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
  20 import org.springframework.kafka.core.KafkaTemplate;
  21 import org.springframework.kafka.core.ProducerFactory;
  22 import org.springframework.kafka.support.SendResult;
  23 import org.springframework.kafka.support.serializer.JsonSerializer;
  24 import org.springframework.kafka.test.EmbeddedKafkaBroker;
  25 import org.springframework.kafka.test.context.EmbeddedKafka;
  26 import org.springframework.kafka.test.utils.KafkaTestUtils;
  27 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  28
  29 import java.time.Clock;
  30 import java.time.ZoneId;
  31 import java.util.HashMap;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.Properties;
  35 import java.util.concurrent.CompletableFuture;
  36
  37 import static de.juplo.kafka.chat.backend.persistence.kafka.KafkaChatHomeTest.TOPIC;
  38
  39
  40 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
  41 @Slf4j
  42 public class KafkaChatHomeTest extends ChatHomeWithShardsTestBase
  43 {
  44   final static String TOPIC = "KAFKA_CHAT_HOME_TEST";
  45
  46   static CompletableFuture<Void> CONSUMER_JOB;
  47
  48
  49   @TestConfiguration
  50   static class Configuration
  51   {
  52     @Bean
  53     KafkaChatHome chatHome(ChatRoomChannel chatRoomChannel)
  54     {
  55       return new KafkaChatHome(numShards(), chatRoomChannel);
  56     }
  57
  58     @Bean
  59     ChatRoomChannel chatRoomChannel(
  60         Producer<String, AbstractMessageTo> chatRoomChannelProducer,
  61         Consumer<String, AbstractMessageTo> chatRoomChannelConsumer,
  62         ZoneId zoneId,
  63         Clock clock)
  64     {
  65       return new ChatRoomChannel(
  66           TOPIC,
  67           chatRoomChannelProducer,
  68           chatRoomChannelConsumer,
  69           zoneId,
  70           numShards(),
  71           8,
  72           clock);
  73     }
  74
  75     @Bean
  76     KafkaTemplate<String, String> kafkaTemplate(EmbeddedKafkaBroker embeddedKafka)
  77     {
  78       Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
  79       ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<>(producerProps);
  80       return new KafkaTemplate<>(producerFactory);
  81     }
  82
  83     @Bean
  84     Producer<String, AbstractMessageTo>  chatRoomChannelProducer(
  85         StringSerializer stringSerializer,
  86         JsonSerializer<AbstractMessageTo> messageSerializer)
  87     {
  88       Map<String, Object> properties = new HashMap<>();
  89       properties.put(
  90           ProducerConfig.CLIENT_ID_CONFIG,
  91           "KAFKACHATHOMETEST_CHATROOM_CHANNEL_PRODUCER");
  92       return new KafkaProducer<>(
  93           properties,
  94           stringSerializer,
  95           messageSerializer);
  96     }
  97
  98     Integer numShards()
  99     {
 100       return 10;
 101     }
 102   }
 103   @BeforeAll
 104   public static void sendAndLoadStoredData(
 105       @Autowired KafkaTemplate<String, String> messageTemplate,
 106       @Autowired Consumer chatRoomChannelConsumer,
 107       @Autowired ThreadPoolTaskExecutor taskExecutor,
 108       @Autowired ChatRoomChannel chatRoomChannel)
 109   {
 110     send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "command_create_chatroom");
 111     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "event_chatmessage_received");
 112     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "event_chatmessage_received");
 113     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "event_chatmessage_received");
 114     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"klaus\", \"text\" : \"Ja? Nein? Vielleicht??\" }", "event_chatmessage_received");
 115
 116     List<TopicPartition> assignedPartitions = List.of(new TopicPartition(TOPIC, 2));
 117     chatRoomChannelConsumer.assign(assignedPartitions);
 118     chatRoomChannel.onPartitionsAssigned(assignedPartitions);
 119     CONSUMER_JOB = taskExecutor
 120         .submitCompletable(chatRoomChannel)
 121         .exceptionally(e ->
 122         {
 123           log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
 124           return null;
 125         });
 126   }
 127
 128   static void send(KafkaTemplate<String, String> kafkaTemplate, String key, String value, String typeId)
 129   {
 130     ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
 131     record.headers().add("__TypeId__", typeId.getBytes());
 132     SendResult<String, String> result = kafkaTemplate.send(record).join();
 133     log.info(
 134         "Sent {}={} to {}",
 135         key,
 136         value,
 137         new TopicPartition(result.getRecordMetadata().topic(), result.getRecordMetadata().partition()));
 138   }
 139
 140   @AfterAll
 141   static void joinConsumerJob(@Autowired Consumer chatRoomChannelConsumer)
 142   {
 143     log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
 144     chatRoomChannelConsumer.wakeup();
 145     log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
 146     CONSUMER_JOB.join();
 147     log.info("Joined the consumer of the ChatRoomChannel");
 148   }
 149 }