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

Author: helltractor

demo-kafka/pom.xml

@@ -49,6 +49,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>

demo-kafka/src/main/resources/application.yml

@@ -4,7 +4,7 @@ spring:
     consumer:
       # set consumer group id (blank for auto)
       group-id: ${KAFKA_CONSUMER_GROUP_ID:}
-      auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:latest}
+      auto-offset-reset: ${KAFKA_CONSUMER_AUTO_OFFSET_RESET:earliest}
       max-poll-records: ${KAFKA_CONSUMER_MAX_POLL_RECORDS:500}
       max-partition-fetch-bytes: ${KAFKA_MAX_PARTITION_FETCH_BYTES:1000000}
 

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

@@ -1,7 +1,6 @@
 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

@@ -1,7 +1,7 @@
 package com.helltractor.demo.messaging;
 
 import com.helltractor.demo.KafkaApplication;
-import com.helltractor.demo.container.ConfluentKafkaContainerCluster;
+import com.helltractor.demo.container.KafkaContainerCluster;
 import com.helltractor.demo.message.TestMessage;
 import com.helltractor.demo.util.IpUtil;
 import org.awaitility.Awaitility;
@@ -12,8 +12,11 @@
 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 org.testcontainers.kafka.KafkaContainer;
 
 import java.time.Duration;
 import java.util.List;
@@ -27,8 +30,11 @@ public final class MessagingFactoryTest {
 
     final static Logger logger = LoggerFactory.getLogger(MessagingFactoryTest.class);
 
+//    @Container
+//    static KafkaContainer kafka = new KafkaContainer("apache/kafka:3.8.0");
+    
     @Container
-    static ConfluentKafkaContainerCluster kafkaCluster = new ConfluentKafkaContainerCluster("7.4.0", 1, 1);
+    static KafkaContainerCluster kafkaCluster = new KafkaContainerCluster("6.2.1", 1, 1);
 
     static class TestConsumer {
 
@@ -48,32 +54,31 @@ int getTotalMessages() {
         }
     }
 
+    @DynamicPropertySource
+    static void dynamicProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.kafka.bootstrap-servers", kafkaCluster::getBootstrapServers);
+    }
+    
     @Autowired
     MessagingFactory messagingFactory;
 
-    MessageProducer<TestMessage> processorOne;
-    MessageProducer<TestMessage> processorTwo;
-    MessageProducer<TestMessage> processorThree;
-    
     @BeforeEach
     void init() {
+//        kafka.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() {
+//        kafka.stop();
         kafkaCluster.stop();
     }
 
     @Test
-    void test() throws InterruptedException {
+    void test() {
+        MessageProducer<TestMessage> processorOne = messagingFactory.createMessageProducer(Messaging.Topic.TOPIC_ONE, TestMessage.class);
+        MessageProducer<TestMessage> processorTwo = messagingFactory.createMessageProducer(Messaging.Topic.TOPIC_TWO, TestMessage.class);
+        MessageProducer<TestMessage> processorThree = messagingFactory.createMessageProducer(Messaging.Topic.TOPIC_THREE, TestMessage.class);
         for (int i = 0; i < 100; i++) {
             TestMessage testMessage = new TestMessage();
             testMessage.message = "Test-" + i;
@@ -90,7 +95,7 @@ void test() throws InterruptedException {
 
         Awaitility.await()
                 .atMost(Duration.ofSeconds(20))
-                .until(() -> testConsumer.getTotalMessages() >= 300);
+                .until(() -> testConsumer.getTotalMessages() == 300);
 
         assertEquals(300, testConsumer.getTotalMessages());
     }