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

Author: helltractor

demo-kafka/pom.xml

@@ -44,6 +44,11 @@
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-testcontainers</artifactId>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka-test</artifactId>
@@ -54,6 +59,12 @@
             <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>${testcontainers.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>kafka</artifactId>

demo-kafka/src/test/java/com/helltractor/demo/ConfluentKafkaContainerStandalone.java

@@ -1,4 +1,4 @@
-package com.helltractor.demo;
+package com.helltractor.demo.container;
 
 import org.apache.kafka.common.Uuid;
 import org.awaitility.Awaitility;

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

@@ -0,0 +1,51 @@
+package com.helltractor.demo.container;
+
+import com.github.dockerjava.api.command.InspectContainerResponse;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
+import org.testcontainers.images.builder.Transferable;
+import org.testcontainers.utility.DockerImageName;
+
+public class KafkaLocalContainer extends GenericContainer<KafkaLocalContainer> {
+    
+    private static final String STARTER_SCRIPT = "/testcontainers_start.sh";
+    
+    public KafkaLocalContainer(String image) {
+        super(DockerImageName.parse(image));
+        withExposedPorts(9092, 8082);
+        var waitStrategy = new WaitAllStrategy().withStrategy(Wait.forLogMessage(".*started.*\\n", 1))
+                .withStrategy(Wait.forHttp("/").forPort(8082).forStatusCode(200));
+        waitingFor(waitStrategy);
+        withCreateContainerCmdModifier(cmd -> {
+            cmd.withEntrypoint("sh");
+        });
+        withCommand("-c", "while [ ! -f " + STARTER_SCRIPT + " ]; do sleep 0.1; done; " + STARTER_SCRIPT);
+    }
+    
+    @Override
+    protected void containerIsStarting(InspectContainerResponse containerInfo) {
+        var defaultListeners = "PLAINTEXT://localhost:29092,CONTROLLER://localhost:29093,PLAINTEXT_HOST://0.0.0.0:9092";
+        var defaultAdvertisedListeners = "PLAINTEXT://localhost:29092,PLAINTEXT_HOST://%s:%d".formatted(getHost(),
+                getMappedPort(9092));
+        var defaultSecurityProtocolMap = "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT";
+        
+        var script = """
+				#!/bin/bash
+				export KAFKA_LISTENERS="%s"
+				export KAFKA_ADVERTISED_LISTENERS="%s"
+				export KAFKA_LISTENER_SECURITY_PROTOCOL_MAP="%s"
+
+				/etc/confluent/docker/run
+				""".formatted(defaultListeners, defaultAdvertisedListeners, defaultSecurityProtocolMap);
+        copyFileToContainer(Transferable.of(script, 0777), STARTER_SCRIPT);
+    }
+    
+    public String getBootstrapServer() {
+        return "%s:%d".formatted(getHost(), getMappedPort(9092));
+    }
+    
+    public String getRestProxyUrl() {
+        return "http://%s:%d".formatted(getHost(), getMappedPort(8082));
+    }
+}

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

@@ -1,69 +1,70 @@
 package com.helltractor.demo.messaging;
 
-import com.helltractor.demo.ConfluentKafkaContainerStandalone;
 import com.helltractor.demo.KafkaApplication;
+import com.helltractor.demo.container.ConfluentKafkaContainerCluster;
 import com.helltractor.demo.message.TestMessage;
 import com.helltractor.demo.util.IpUtil;
+import org.awaitility.Awaitility;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.context.annotation.Lazy;
-import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 @SpringBootTest(classes = KafkaApplication.class)
-@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
 public class MessagingTest {
 
     final static Logger logger = LoggerFactory.getLogger(MessagingTest.class);
 
-    @Lazy
+    static class TestConsumer {
+        
+        final AtomicInteger messageCount = new AtomicInteger();
+        
+        final long startTime = System.currentTimeMillis();
+        
+        void processMessages(List<TestMessage> messages) {
+            messageCount.addAndGet(messages.size());
+            long currentTime = System.currentTimeMillis();
+            logger.info("Received {} messages, Total: {}, Elapsed time: {} ms",
+                    messages.size(), messageCount.get(), currentTime - startTime);
+        }
+        
+        int getTotalMessages() {
+            return messageCount.get();
+        }
+    }
+    
     @Autowired
     MessagingFactory messagingFactory;
 
-    ConfluentKafkaContainerStandalone standalone;
+    ConfluentKafkaContainerCluster cluster;
 
     MessageProducer<TestMessage> processorOne;
     MessageProducer<TestMessage> processorTwo;
     MessageProducer<TestMessage> processorThree;
 
     @BeforeEach
     void init() {
-        standalone = new ConfluentKafkaContainerStandalone("7.4.0");
+        cluster = new ConfluentKafkaContainerCluster("7.4.0", 1, 1);
+        cluster.start();
         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() {
-        standalone.stop();
-    }
-    
-    static class TestConsumer {
-        
-        final AtomicInteger messageCount = new AtomicInteger();
-        
-        final long startTime = System.currentTimeMillis();
-        
-        void processMessages(List<TestMessage> messages) {
-            messageCount.addAndGet(messages.size());
-            long currentTime = System.currentTimeMillis();
-            logger.info("Received {} messages, Total: {}, Elapsed time: {} ms",
-                    messages.size(), messageCount.get(), currentTime - startTime);
-        }
-        
-        int getTotalMessages() {
-            return messageCount.get();
-        }
+        cluster.stop();
     }
 
     @Test
@@ -82,7 +83,10 @@ void test() throws InterruptedException {
             messagingFactory.createBatchMessageListener(topic, groupId, testConsumer::processMessages);
         }
 
-        Thread.sleep(10000);
+        Awaitility.await()
+                .atMost(Duration.ofSeconds(10))
+                .until(() -> testConsumer.getTotalMessages() == 300);
+        
         assertEquals(300, testConsumer.getTotalMessages());
     }
 }