Add Kafka tests with Testcontainers

Author: kasramp

pom.xml

@@ -54,6 +54,24 @@
 			<artifactId>spring-kafka-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>testcontainers</artifactId>
+			<version>1.17.3</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>junit-jupiter</artifactId>
+			<version>1.17.3</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.testcontainers</groupId>
+			<artifactId>kafka</artifactId>
+			<version>1.17.3</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 
 	<build>

src/main/java/com/madadipouya/springkafkatest/kafka/consumer/UserKafkaConsumer.java

@@ -1,6 +1,7 @@
 package com.madadipouya.springkafkatest.kafka.consumer;
 
 import com.madadipouya.springkafkatest.dto.User;
+import com.madadipouya.springkafkatest.service.UserService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.kafka.annotation.KafkaListener;
@@ -12,7 +13,13 @@
 @Component
 public class UserKafkaConsumer {
 
-    private final Logger logger = LoggerFactory.getLogger(UserKafkaConsumer.class);
+    private static final Logger logger = LoggerFactory.getLogger(UserKafkaConsumer.class);
+
+    private final UserService userService;
+
+    public UserKafkaConsumer(UserService userService) {
+        this.userService = userService;
+    }
 
     @KafkaListener(topics = "${spring.kafka.topic.name}",
             concurrency = "${spring.kafka.consumer.level.concurrency:3}")
@@ -21,6 +28,7 @@ public void logKafkaMessages(@Payload User user,
                                  @Header(KafkaHeaders.RECEIVED_PARTITION_ID) Integer partition,
                                  @Header(KafkaHeaders.OFFSET) Long offset) {
         logger.info("Received a message contains a user information with id {}, from {} topic, " +
-                        "{} partition, and {} offset", user.getUuid(), topic, partition, offset);
+                "{} partition, and {} offset", user.getUuid(), topic, partition, offset);
+        userService.save(user);
     }
 }

src/main/java/com/madadipouya/springkafkatest/service/UserService.java

@@ -0,0 +1,8 @@
+package com.madadipouya.springkafkatest.service;
+
+import com.madadipouya.springkafkatest.dto.User;
+
+public interface UserService {
+
+    void save(User user);
+}

src/main/java/com/madadipouya/springkafkatest/service/impl/DefaultUserService.java

@@ -0,0 +1,18 @@
+package com.madadipouya.springkafkatest.service.impl;
+
+import com.madadipouya.springkafkatest.dto.User;
+import com.madadipouya.springkafkatest.service.UserService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+@Service
+public class DefaultUserService implements UserService {
+
+    private static final Logger logger = LoggerFactory.getLogger(DefaultUserService.class);
+
+    @Override
+    public void save(User user) {
+        logger.info("Saving user with id = {}", user.getUuid());
+    }
+}

src/test/java/com/madadipouya/springkafkatest/kafka/UserKafkaTestcontainersTest.java

@@ -0,0 +1,58 @@
+package com.madadipouya.springkafkatest.kafka;
+
+import com.madadipouya.springkafkatest.dto.User;
+import com.madadipouya.springkafkatest.kafka.consumer.UserKafkaConsumer;
+import com.madadipouya.springkafkatest.kafka.producer.UserKafkaProducer;
+import com.madadipouya.springkafkatest.service.UserService;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+@Testcontainers
+@SpringBootTest
+class UserKafkaTestcontainersTest {
+
+    @Container
+    static KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:latest"));
+
+    @DynamicPropertySource
+    static void kafkaProperties(DynamicPropertyRegistry registry) {
+        registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
+    }
+
+    @Autowired
+    private UserKafkaProducer userKafkaProducer;
+
+    @Autowired
+    private UserKafkaConsumer userKafkaConsumer;
+
+    @MockBean
+    private UserService userService;
+
+    @Test
+    void testProduceAndConsumeKafkaMessage() {
+        ArgumentCaptor<User> captor = ArgumentCaptor.forClass(User.class);
+        User user = new User("11111", "John", "Wick");
+
+        userKafkaProducer.writeToKafka(user);
+
+        verify(userService, timeout(5000)).save(captor.capture());
+        assertNotNull(captor.getValue());
+        assertEquals("11111", captor.getValue().getUuid());
+        assertEquals("John", captor.getValue().getFirstName());
+        assertEquals("Wick", captor.getValue().getLastName());
+    }
+}

src/test/java/com/madadipouya/springkafkatest/kafka/consumer/UserKafkaConsumerTest.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.madadipouya.springkafkatest.dto.User;
+import com.madadipouya.springkafkatest.service.UserService;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -14,6 +15,7 @@
 import org.mockito.Captor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
 import org.springframework.boot.test.mock.mockito.SpyBean;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
@@ -25,6 +27,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.verify;
 
@@ -46,6 +49,9 @@ class UserKafkaConsumerTest {
     @SpyBean
     private UserKafkaConsumer userKafkaConsumer;
 
+    @MockBean
+    private UserService userService;
+
     @Captor
     ArgumentCaptor<User> userArgumentCaptor;
 
@@ -85,6 +91,7 @@ void testLogKafkaMessages() throws JsonProcessingException {
         assertEquals(TOPIC_NAME, topicArgumentCaptor.getValue());
         assertEquals(0, partitionArgumentCaptor.getValue());
         assertEquals(0, offsetArgumentCaptor.getValue());
+        verify(userService).save(any(User.class));
     }
 
     @AfterAll