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

Author: helltractor

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

@@ -0,0 +1,42 @@
+package com.helltractor.demo;
+
+import org.testcontainers.containers.Network;
+import org.testcontainers.kafka.ConfluentKafkaContainer;
+import org.testcontainers.lifecycle.Startable;
+import org.testcontainers.utility.DockerImageName;
+
+public class ConfluentKafkaContainerStandalone implements Startable {
+    
+    private final Network network;
+    
+    private final ConfluentKafkaContainer container;
+    
+    public ConfluentKafkaContainerStandalone(String confluentPlatformVersion) {
+        this.network = Network.newNetwork();
+        this.container = new ConfluentKafkaContainer(
+                DockerImageName.parse("confluentinc/cp-kafka").withTag(confluentPlatformVersion)
+        )
+                .withNetwork(this.network)
+                .withNetworkAliases("kafka-broker")
+                .withEnv("KAFKA_BROKER_ID", "1")
+                .withEnv("KAFKA_NODE_ID", "1")
+                .withEnv("KAFKA_LISTENERS", "PLAINTEXT://0.0.0.0:9092")
+                .withEnv("KAFKA_ADVERTISED_LISTENERS", "PLAINTEXT://localhost:9092")
+                .withEnv("KAFKA_CONTROLLER_QUORUM_VOTERS", "1@kafka-broker:9093")
+                .withEnv("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "true")
+                .withEnv("KAFKA_LOG_DIRS", "/var/lib/kafka/data")
+                .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1")
+                .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1");
+    }
+    
+    
+    @Override
+    public void start() {
+        container.start();
+    }
+    
+    @Override
+    public void stop() {
+        container.stop();
+    }
+}

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

@@ -1,6 +1,6 @@
 package com.helltractor.demo.messaging;
 
-import com.helltractor.demo.ConfluentKafkaContainerCluster;
+import com.helltractor.demo.ConfluentKafkaContainerStandalone;
 import com.helltractor.demo.KafkaApplication;
 import com.helltractor.demo.message.TestMessage;
 import com.helltractor.demo.util.IpUtil;
@@ -11,38 +11,41 @@
 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 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
     @Autowired
     MessagingFactory messagingFactory;
 
-    ConfluentKafkaContainerCluster cluster;
+    ConfluentKafkaContainerStandalone standalone;
 
     MessageProducer<TestMessage> processorOne;
     MessageProducer<TestMessage> processorTwo;
     MessageProducer<TestMessage> processorThree;
 
     @BeforeEach
     void init() {
-        cluster = new ConfluentKafkaContainerCluster("7.4.0", 1, 1);
-        cluster.start();
+        standalone = new ConfluentKafkaContainerStandalone("7.4.0");
         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();
+        standalone.stop();
     }
 
     static class TestConsumer {
@@ -65,7 +68,7 @@ int getTotalMessages() {
 
     @Test
     void test() throws InterruptedException {
-        for (int i = 0; i < 1000; i++) {
+        for (int i = 0; i < 100; i++) {
             TestMessage testMessage = new TestMessage();
             testMessage.message = "Test-" + i;
             processorOne.sendMessages(testMessage);
@@ -80,6 +83,6 @@ void test() throws InterruptedException {
         }
 
         Thread.sleep(10000);
-        assertEquals(3000, testConsumer.getTotalMessages());
+        assertEquals(300, testConsumer.getTotalMessages());
     }
 }