From 77acedd8b72f4c91bd265ff53604828af9703b29 Mon Sep 17 00:00:00 2001
From: Kai Moritz <kai@juplo.de>
Date: Sun, 20 Aug 2023 11:12:55 +0200
Subject: [PATCH] WIP

---
 .../chat/backend/KafkaConfigurationIT.java    | 39 +++++++++++++++++--
 1 file changed, 35 insertions(+), 4 deletions(-)

diff --git a/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java b/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java
index 7b2df45a..f25feb69 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/KafkaConfigurationIT.java
@@ -1,19 +1,24 @@
 package de.juplo.kafka.chat.backend;
 
 import de.juplo.kafka.chat.backend.domain.ShardingStrategy;
+import de.juplo.kafka.chat.backend.persistence.kafka.ChatRoomChannel;
 import de.juplo.kafka.chat.backend.persistence.kafka.KafkaServicesApplicationRunner;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 
 import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.TOPIC;
 
@@ -29,19 +34,33 @@ import static de.juplo.kafka.chat.backend.KafkaConfigurationIT.TOPIC;
         "chat.backend.kafka.num-partitions=10",
         })
 @EmbeddedKafka(topics = { TOPIC }, partitions = 10)
+@Slf4j
 class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
 {
   final static String TOPIC = "TEST_CHATROOM_CHANNEL";
 
+  static CompletableFuture<Void> CONSUMER_JOB;
+
   @MockBean
   KafkaServicesApplicationRunner kafkaServicesApplicationRunner;
 
 
   @BeforeAll
   public static void prepareChatRoomChannelConsumer(
-      @Autowired Consumer chatRoomChannelConsumer)
+      @Autowired Consumer chatRoomChannelConsumer,
+      @Autowired ThreadPoolTaskExecutor taskExecutor,
+  @Autowired ChatRoomChannel chatRoomChannel)
   {
-    chatRoomChannelConsumer.assign(List.of(new TopicPartition(TOPIC, 6)));
+    List<TopicPartition> assignedPartitions = List.of(new TopicPartition(TOPIC, 2));
+    chatRoomChannelConsumer.assign(assignedPartitions);
+    CONSUMER_JOB = taskExecutor
+        .submitCompletable(chatRoomChannel)
+        .exceptionally(e ->
+        {
+          log.error("The consumer for the ChatRoomChannel exited abnormally!", e);
+          return null;
+        });
+    chatRoomChannel.onPartitionsAssigned(assignedPartitions);
   }
 
   @BeforeAll
@@ -51,7 +70,7 @@ class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
   {
     UUID chatRoomId = UUID.fromString("5c73531c-6fc4-426c-adcb-afc5c140a0f7");
     int shard = shardingStrategy.selectShard(chatRoomId);
-    send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": " + shard + ", \"name\": \"FOO\" }", "create");
+    send(messageTemplate, "5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\": \"5c73531c-6fc4-426c-adcb-afc5c140a0f7\", \"shard\": 2, \"name\": \"FOO\" }", "create");
     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"peter\", \"text\" : \"Hallo, ich heiße Peter!\" }", "message");
     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 1, \"user\" : \"ute\", \"text\" : \"Ich bin Ute...\" }", "message");
     send(messageTemplate,"5c73531c-6fc4-426c-adcb-afc5c140a0f7","{ \"id\" : 2, \"user\" : \"peter\", \"text\" : \"Willst du mit mir gehen?\" }", "message");
@@ -62,6 +81,18 @@ class KafkaConfigurationIT extends AbstractConfigurationWithShardingIT
   {
     ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC, key, value);
     record.headers().add("__TypeId__", typeId.getBytes());
-    kafkaTemplate.send(record);
+    kafkaTemplate
+        .send(record)
+        .thenAccept(result -> log.info("Sent {}={} to partition {}", key, value, result.getRecordMetadata().partition()));
+  }
+
+  @AfterAll
+  static void joinConsumerJob(@Autowired Consumer chatRoomChannelConsumer)
+  {
+    log.info("Signaling the consumer of the CahtRoomChannel to quit its work");
+    chatRoomChannelConsumer.wakeup();
+    log.info("Waiting for the consumer of the ChatRoomChannel to finish its work");
+    CONSUMER_JOB.join();
+    log.info("Joined the consumer of the ChatRoomChannel");
   }
 }
-- 
2.20.1