From: Kai Moritz <kai@juplo.de>
Date: Thu, 7 Mar 2024 16:43:30 +0000 (+0100)
Subject: test: HandoverIT-POC - Refactored the startup of backend-containers
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=fb8d39e2a3ff7d6d93aebdb0697ac50b911075b7;p=demos%2Fkafka%2Fchat

test: HandoverIT-POC - Refactored the startup of backend-containers

* The backend-containers are explicitly started during the test.
* When a backend is started, it is waited for, that the started backend
  reportes its status as `UP`, _and_, that alle writers are again able to
  send messages, afterwards.
---

diff --git a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
index 3c135591..f9586473 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverIT.java
@@ -37,6 +37,10 @@ public abstract class AbstractHandoverIT
   @Test
   void test() throws InterruptedException
   {
+    log.info("Starting backend-1...");
+    containers.startBackend(containers.backend1, new TestWriter[0]);
+    log.info("backend-1 started!");
+
     ChatRoomInfoTo[] chatRooms = Flux
         .range(0, NUM_CHATROOMS)
         .flatMap(i -> createChatRoom("room-" + i))
diff --git a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverITContainers.java b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverITContainers.java
index 4e19defd..e94ecc61 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverITContainers.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/AbstractHandoverITContainers.java
@@ -45,15 +45,24 @@ public abstract class AbstractHandoverITContainers
   {
     setUpExtra();
     haproxy.start();
-    backend1.start();
-    // backend2.start();
-    // backend3.start();
+  }
+
+  void startBackend(
+      GenericContainer backend,
+      TestWriter[] testWriters)
+  {
+    backend.start();
+
+    int[] numSentMessages = Arrays
+        .stream(testWriters)
+        .mapToInt(testWriter -> testWriter.getNumSentMessages())
+        .toArray();
 
     Awaitility
         .await()
-        .atMost(Duration.ofMinutes(10))
+        .atMost(Duration.ofSeconds(30))
         .until(() -> WebClient
-            .create("http://localhost:" + backend1.getMappedPort(8080))
+            .create("http://localhost:" + backend.getMappedPort(8080))
             .get()
             .uri("/actuator/health")
             .exchangeToMono(response ->
@@ -78,10 +87,9 @@ public abstract class AbstractHandoverITContainers
         .withSignal("HUP")
         .exec();
 
-
     Awaitility
         .await()
-        .atMost(Duration.ofMinutes(10))
+        .atMost(Duration.ofSeconds(30))
         .until(() -> WebClient
             .create("http://localhost:" + haproxy.getMappedPort(8400))
             .get()
@@ -101,6 +109,29 @@ public abstract class AbstractHandoverITContainers
               }
             })
             .block());
+
+    Awaitility
+        .await()
+        .atMost(Duration.ofSeconds(30))
+        .until(() ->
+        {
+          for (int i = 0; i < testWriters.length; i++)
+          {
+            TestWriter testWriter = testWriters[i];
+            int sentTotal = testWriter.getNumSentMessages();
+            if (sentTotal == numSentMessages[i])
+            {
+              log.info(
+                  "No progress for {}: sent-before={}, sent-total={}",
+                  testWriter,
+                  numSentMessages[i],
+                  sentTotal);
+              return false;
+            }
+          }
+
+          return true;
+        });
   }
 
   abstract String[] getBackendCommand();
diff --git a/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java b/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
index ea6ddfe3..a6ac5696 100644
--- a/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
+++ b/src/test/java/de/juplo/kafka/chat/backend/TestWriter.java
@@ -2,6 +2,7 @@ package de.juplo.kafka.chat.backend;
 
 import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo;
 import de.juplo.kafka.chat.backend.api.MessageTo;
+import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -44,9 +45,10 @@ public class TestWriter
         .delayElements(Duration.ofMillis(ThreadLocalRandom.current().nextLong(500, 1500)))
         .map(i -> "Message #" + i)
         .flatMap(message -> sendMessage(chatRoom, message)
-            .retryWhen(Retry.fixedDelay(10, Duration.ofSeconds(1))))
+            .retryWhen(Retry.fixedDelay(30, Duration.ofSeconds(1))))
         .doOnNext(message ->
         {
+          numSentMessages++;
           sentMessages.add(message);
           log.info(
               "{} sent a message to {}: {}",
@@ -102,6 +104,8 @@ public class TestWriter
   final List<MessageTo> sentMessages = new LinkedList<>();
 
   volatile boolean running = true;
+  @Getter
+  private volatile int numSentMessages = 0;
 
 
   TestWriter(Integer port, ChatRoomInfoTo chatRoom, String username)