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;
@@ -39,7 +40,7 @@ public ConfluentKafkaContainerCluster(String confluentPlatformVersion, int broke
 
         String controllerQuorumVoters = IntStream
                 .range(0, brokersNum)
-                .mapToObj(brokerNum -> String.format("%d@broker-%d:9092", brokerNum, brokerNum))
+                .mapToObj(brokerNum -> String.format("%d@broker-%d:9094", brokerNum, brokerNum))
                 .collect(Collectors.joining(","));
 
         String clusterId = Uuid.randomUuid().toString();

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

@@ -12,6 +12,10 @@
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
 
 import java.time.Duration;
 import java.util.List;
@@ -20,10 +24,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();
@@ -42,31 +50,33 @@ int getTotalMessages() {
         }
     }
 
+    @DynamicPropertySource
+    static void kafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.kafka.bootstrap-servers", kafkaCluster::getBootstrapServers);
+    }
+    
     @Autowired
     MessagingFactory messagingFactory;
 
-    ConfluentKafkaContainerCluster cluster;
-    
     MessageProducer<TestMessage> processorOne;
     MessageProducer<TestMessage> processorTwo;
     MessageProducer<TestMessage> processorThree;
 
     @BeforeEach
     void init() {
-        cluster = new ConfluentKafkaContainerCluster("7.4.0", 3, 2);
-        cluster.start();
+        kafkaCluster.start();
         Awaitility.await()
                 .atMost(Duration.ofMinutes(1))
                 .pollInterval(Duration.ofSeconds(5))
-                .until(() -> cluster.getBrokers().stream().allMatch(b -> b.isRunning()));
+                .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
@@ -86,8 +96,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());
     }