chore(demo-kafka): fix org.apache.kafka.common.errors.TimeoutException

Author: helltractor

demo-kafka/pom.xml

@@ -65,6 +65,12 @@
             <version>${testcontainers.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

demo-kafka/src/test/java/com/helltractor/demo/container/ConfluentKafkaContainerCluster.java

@@ -1,6 +1,7 @@
 package com.helltractor.demo.container;
 
 import org.apache.kafka.common.Uuid;
+import org.apache.kafka.clients.NetworkClient;
 import org.awaitility.Awaitility;
 import org.testcontainers.containers.Container;
 import org.testcontainers.containers.GenericContainer;

demo-kafka/src/test/java/com/helltractor/demo/messaging/MessagingFactoryTest.java

@@ -12,6 +12,8 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.time.Duration;
 import java.util.List;
@@ -20,10 +22,14 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 @SpringBootTest(classes = KafkaApplication.class)
-public class MessagingFactoryTest {
+@Testcontainers
+public final class MessagingFactoryTest {
 
     final static Logger logger = LoggerFactory.getLogger(MessagingFactoryTest.class);
 
+    @Container
+    static ConfluentKafkaContainerCluster kafkaCluster = new ConfluentKafkaContainerCluster("7.4.0", 1, 1);
+    
     static class TestConsumer {
 
         final AtomicInteger messageCount = new AtomicInteger();
@@ -45,24 +51,25 @@ int getTotalMessages() {
     @Autowired
     MessagingFactory messagingFactory;
 
-    ConfluentKafkaContainerCluster cluster;
-    
     MessageProducer<TestMessage> processorOne;
     MessageProducer<TestMessage> processorTwo;
     MessageProducer<TestMessage> processorThree;
 
     @BeforeEach
     void init() {
-        cluster = new ConfluentKafkaContainerCluster("7.4.0", 1, 1);
-        cluster.start();
+        kafkaCluster.start();
+        Awaitility.await()
+                .atMost(Duration.ofMinutes(1))
+                .pollInterval(Duration.ofSeconds(5))
+                .until(() -> kafkaCluster.getBrokers().stream().allMatch(b -> b.isRunning()));
         processorOne = messagingFactory.createMessageProducer(Messaging.Topic.TOPIC_ONE, TestMessage.class);
         processorTwo = messagingFactory.createMessageProducer(Messaging.Topic.TOPIC_TWO, TestMessage.class);
         processorThree = messagingFactory.createMessageProducer(Messaging.Topic.TOPIC_THREE, TestMessage.class);
     }
 
     @AfterEach
     void destroy() {
-        cluster.stop();
+        kafkaCluster.stop();
     }
 
     @Test
@@ -82,8 +89,8 @@ void test() throws InterruptedException {
         }
 
         Awaitility.await()
-                .atMost(Duration.ofSeconds(10))
-                .until(() -> testConsumer.getTotalMessages() == 300);
+                .atMost(Duration.ofSeconds(20))
+                .until(() -> testConsumer.getTotalMessages() >= 300);
 
         assertEquals(300, testConsumer.getTotalMessages());
     }