Add kafka container

README.md

@@ -24,11 +24,12 @@ Spring-TestContainers provides out-of-the-box support for the following containe
 
 Spring-TestContainers provides out-of-the-box support for the following containers. You can enable each one via a dedicated annotation in your test classes:
 
-| Container       | Annotation                        | Example Usage                                        | Notes                          |
-|----------------|-----------------------------------|-----------------------------------------------------|--------------------------------|
-| **PostgreSQL** | `@EnablePostgresContainer`        | `@EnablePostgresContainer(version = "15")`          | Uses `PostgreSQLContainer`     |
-| **MySQL**      | `@EnableMySQLContainer`           | `@EnableMySQLContainer(version = "8")`              | Uses `MySQLContainer`          |
-| **Ollama (AI)**| `@EnableOllamaContainer`          | `@EnableOllamaContainer(model = "llama2")`          | Starts Ollama with auto-pull   |
+| Container  | Annotation                 | Example Usage                              | Notes                        |
+|------------|----------------------------|--------------------------------------------|------------------------------|
+| **PostgreSQL** | `@EnablePostgresContainer` | `@EnablePostgresContainer(version = "15")` | Uses `PostgreSQLContainer`   |
+| **MySQL**  | `@EnableMySQLContainer`    | `@EnableMySQLContainer(version = "8")`     | Uses `MySQLContainer`        |
+| **Kafka**  | `@EnableKafkaContainer`    | `@EnableKafkaContainer(version = "3.9.1")` | Use `KafkaContainer`         |
+| **Ollama (AI)** | `@EnableOllamaContainer`   | `@EnableOllamaContainer(model = "llama2")` | Starts Ollama with auto-pull |
 
 
 ## Comparison: TestContainers with Spring vs Spring-TestContainers
@@ -152,6 +153,7 @@ Add the core library along with the database module(s) you plan to use. Each dat
 // Add one or more of the following database modules
 testImplementation("io.flowinquiry.testcontainers:postgresql:<!-- Replace with the latest version -->") // PostgreSQL support
 testImplementation("io.flowinquiry.testcontainers:mysql:<!-- Replace with the latest version -->")      // MySQL support
+testImplementation("io.flowinquiry.testcontainers:kafka:<!-- Replace with the latest version -->")
 testImplementation("io.flowinquiry.testcontainers:ollama:<!-- Replace with the latest version -->")     // Ollama support 
 ```
 
@@ -178,6 +180,14 @@ testImplementation("io.flowinquiry.testcontainers:ollama:<!-- Replace with the l
     <scope>test</scope>
 </dependency>
 
+<!-- Add this dependency to test Kafka -->
+<dependency>
+    <groupId>io.flowinquiry.testcontainers</groupId>
+    <artifactId>kafka</artifactId>
+    <version><!-- Replace with the latest version --></version>
+    <scope>test</scope>
+</dependency>
+
 <!-- Add this dependency to test Ollama container -->
 <dependency>
     <groupId>io.flowinquiry.testcontainers</groupId>
@@ -228,6 +238,7 @@ Currently, the following containers are supported:
 
 - PostgreSQL
 - MySQL
+- Kafka
 - Ollama
 
 ## Documentation
@@ -253,6 +264,12 @@ The project includes several example modules demonstrating how to use Spring-Tes
 
 These examples provide a good starting point for integrating Spring-TestContainers into your own projects.
 
+### [springboot-kafka](examples/springboot-kafka)
+
+* Spring Boot applications having Kafka producer and consumer
+
+* Show how to integrate kafka container with minimal configuration
+
 ### [springboot-ollama](examples/springboot-ollama)
 
 * Spring Boot applications using Spring AI and Ollama

examples/springboot-kafka/build.gradle.kts

@@ -0,0 +1,34 @@
+plugins {
+    id("buildlogic.java-application-conventions")
+    alias(libs.plugins.spring.dependency.management)
+}
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    implementation(project(":spring-testcontainers"))
+    implementation(project(":modules:kafka"))
+    implementation(platform(libs.spring.bom))
+    implementation(platform(libs.spring.boot.bom))
+    implementation(libs.slf4j.api)
+    implementation(libs.logback.classic)
+    implementation(libs.spring.boot.starter)
+    implementation("org.springframework.kafka:spring-kafka")
+    implementation(libs.spring.boot.autoconfigure)
+    implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
+    testImplementation(platform(libs.junit.bom))
+    testImplementation(libs.junit.jupiter)
+    testImplementation(libs.junit.platform.launcher)
+    testImplementation(libs.spring.boot.starter.test)
+    testImplementation("org.springframework.kafka:spring-kafka-test")
+}
+
+tasks.test {
+    useJUnitPlatform()
+}
+
+application {
+    mainClass.set("io.flowinquiry.testcontainers.examples.kafka.KafkaDemoApp")
+}

examples/springboot-kafka/src/main/java/io/flowinquiry/testcontainers/examples/kafka/KafkaDemoApp.java

@@ -0,0 +1,18 @@
+package io.flowinquiry.testcontainers.examples.kafka;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.kafka.annotation.EnableKafka;
+
+/**
+ * Spring Boot application for the Kafka demo. This application demonstrates how to use Kafka with
+ * Spring Boot and Testcontainers.
+ */
+@SpringBootApplication
+@EnableKafka
+public class KafkaDemoApp {
+
+  public static void main(String[] args) {
+    SpringApplication.run(KafkaDemoApp.class, args);
+  }
+}

examples/springboot-kafka/src/main/java/io/flowinquiry/testcontainers/examples/kafka/config/KafkaConfig.java

@@ -0,0 +1,23 @@
+package io.flowinquiry.testcontainers.examples.kafka.config;
+
+import org.apache.kafka.clients.admin.NewTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.TopicBuilder;
+
+/** Configuration class for Kafka. Defines the Kafka topic and other Kafka-related beans. */
+@Configuration
+public class KafkaConfig {
+
+  public static final String TOPIC_NAME = "test-topic";
+
+  /**
+   * Creates a Kafka topic.
+   *
+   * @return the Kafka topic
+   */
+  @Bean
+  public NewTopic testTopic() {
+    return TopicBuilder.name(TOPIC_NAME).partitions(1).replicas(1).build();
+  }
+}

examples/springboot-kafka/src/main/java/io/flowinquiry/testcontainers/examples/kafka/consumer/MessageConsumer.java

@@ -0,0 +1,60 @@
+package io.flowinquiry.testcontainers.examples.kafka.consumer;
+
+import io.flowinquiry.testcontainers.examples.kafka.config.KafkaConfig;
+import io.flowinquiry.testcontainers.examples.kafka.model.Message;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Service;
+
+/** Service for consuming messages from Kafka. */
+@Service
+public class MessageConsumer {
+
+  private static final Logger log = LoggerFactory.getLogger(MessageConsumer.class);
+
+  private final List<Message> receivedMessages = new ArrayList<>();
+  private CountDownLatch latch = new CountDownLatch(1);
+
+  /**
+   * Receives messages from the Kafka topic.
+   *
+   * @param message the received message
+   */
+  @KafkaListener(topics = KafkaConfig.TOPIC_NAME, groupId = "test-group")
+  public void receive(Message message) {
+    log.info("Received message: {}", message);
+    receivedMessages.add(message);
+    latch.countDown();
+  }
+
+  /**
+   * Gets the list of received messages.
+   *
+   * @return the list of received messages
+   */
+  public List<Message> getReceivedMessages() {
+    return new ArrayList<>(receivedMessages);
+  }
+
+  /**
+   * Gets the latch that counts down when a message is received.
+   *
+   * @return the latch
+   */
+  public CountDownLatch getLatch() {
+    return latch;
+  }
+
+  /**
+   * Resets the latch to the specified count.
+   *
+   * @param count the count to reset the latch to
+   */
+  public void resetLatch(int count) {
+    latch = new CountDownLatch(count);
+  }
+}

examples/springboot-kafka/src/main/java/io/flowinquiry/testcontainers/examples/kafka/model/Message.java

@@ -0,0 +1,39 @@
+package io.flowinquiry.testcontainers.examples.kafka.model;
+
+import java.time.LocalDateTime;
+
+/** A simple message model class for Kafka messages. */
+public class Message {
+
+  private String content;
+  private LocalDateTime timestamp;
+
+  // Default constructor required for JSON deserialization
+  public Message() {}
+
+  public Message(String content) {
+    this.content = content;
+    this.timestamp = LocalDateTime.now();
+  }
+
+  public String getContent() {
+    return content;
+  }
+
+  public void setContent(String content) {
+    this.content = content;
+  }
+
+  public LocalDateTime getTimestamp() {
+    return timestamp;
+  }
+
+  public void setTimestamp(LocalDateTime timestamp) {
+    this.timestamp = timestamp;
+  }
+
+  @Override
+  public String toString() {
+    return "Message{" + "content='" + content + '\'' + ", timestamp=" + timestamp + '}';
+  }
+}

examples/springboot-kafka/src/main/java/io/flowinquiry/testcontainers/examples/kafka/producer/MessageProducer.java

@@ -0,0 +1,47 @@
+package io.flowinquiry.testcontainers.examples.kafka.producer;
+
+import io.flowinquiry.testcontainers.examples.kafka.config.KafkaConfig;
+import io.flowinquiry.testcontainers.examples.kafka.model.Message;
+import java.util.concurrent.CompletableFuture;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Service;
+
+/** Service for producing messages to Kafka. */
+@Service
+public class MessageProducer {
+
+  private static final Logger log = LoggerFactory.getLogger(MessageProducer.class);
+
+  private final KafkaTemplate<String, Message> kafkaTemplate;
+
+  public MessageProducer(KafkaTemplate<String, Message> kafkaTemplate) {
+    this.kafkaTemplate = kafkaTemplate;
+  }
+
+  /**
+   * Sends a message to the Kafka topic.
+   *
+   * @param message the message to send
+   * @return a CompletableFuture that will be completed when the send operation completes
+   */
+  public CompletableFuture<SendResult<String, Message>> sendMessage(Message message) {
+    log.info("Sending message: {}", message);
+    CompletableFuture<SendResult<String, Message>> future =
+        kafkaTemplate.send(KafkaConfig.TOPIC_NAME, message);
+
+    future.whenComplete(
+        (result, ex) -> {
+          if (ex == null) {
+            log.info("Message sent successfully: {}", message);
+            log.info("Offset: {}", result.getRecordMetadata().offset());
+          } else {
+            log.error("Failed to send message: {}", message, ex);
+          }
+        });
+
+    return future;
+  }
+}

examples/springboot-kafka/src/main/resources/application.yml

@@ -0,0 +1,10 @@
+spring:
+  kafka:
+    producer:
+      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
+      key-serializer: org.apache.kafka.common.serialization.StringSerializer
+    consumer:
+      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      properties:
+        spring.json.trusted.packages: io.flowinquiry.testcontainers.examples.kafka.model

examples/springboot-kafka/src/test/java/io/flowinquiry/testcontainers/examples/kafka/KafkaDemoAppTest.java

@@ -0,0 +1,58 @@
+package io.flowinquiry.testcontainers.examples.kafka;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.flowinquiry.testcontainers.examples.kafka.consumer.MessageConsumer;
+import io.flowinquiry.testcontainers.examples.kafka.model.Message;
+import io.flowinquiry.testcontainers.examples.kafka.producer.MessageProducer;
+import io.flowinquiry.testcontainers.kafka.EnableKafkaContainer;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+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;
+
+@SpringBootTest
+@EnableKafkaContainer
+public class KafkaDemoAppTest {
+
+  private static final Logger log = LoggerFactory.getLogger(KafkaDemoAppTest.class);
+
+  @Autowired private MessageProducer producer;
+  @Autowired private MessageConsumer consumer;
+
+  @Test
+  public void testKafkaMessaging() throws Exception {
+    // Create a test message
+    Message message = new Message("Test message content");
+
+    // Send the message
+    log.info("Sending test message");
+    producer.sendMessage(message);
+
+    // Wait for the consumer to receive the message
+    log.info("Waiting for consumer to receive the message");
+    boolean messageReceived = consumer.getLatch().await(10, TimeUnit.SECONDS);
+
+    // Verify that the message was received
+    assertTrue(messageReceived, "Message should be received within timeout");
+
+    // Get the received messages
+    List<Message> receivedMessages = consumer.getReceivedMessages();
+
+    // Verify that exactly one message was received
+    assertEquals(1, receivedMessages.size(), "Should receive exactly one message");
+
+    // Verify the content of the received message
+    Message receivedMessage = receivedMessages.get(0);
+    assertNotNull(receivedMessage, "Received message should not be null");
+    assertEquals(
+        "Test message content", receivedMessage.getContent(), "Message content should match");
+
+    log.info("Test completed successfully");
+  }
+}

gradle.properties

@@ -2,5 +2,5 @@
 # https://docs.gradle.org/current/userguide/build_environment.html#sec:gradle_configuration_properties
 
 org.gradle.configuration-cache=true
-version=0.9.2
+version=0.9.3
 

gradle/libs.versions.toml

@@ -8,7 +8,7 @@ spotless = "7.0.3"
 logback = "1.5.18"
 slf4j = "2.0.17"
 spring = "6.2.7"
-spring-boot = "3.5.0"
+spring-boot = "3.5.3"
 spring-dependency-management="1.1.7"
 postgresql = "42.7.2"
 mysql="8.0.33"
@@ -29,6 +29,7 @@ testcontainers-jdbc = { group = "org.testcontainers", name = "jdbc", version.ref
 testcontainers-mysql = { group = "org.testcontainers", name = "mysql", version.ref = "testcontainers" }
 testcontainers-postgresql = { group = "org.testcontainers", name = "postgresql", version.ref = "testcontainers" }
 testcontainers-ollama = { group = "org.testcontainers", name = "ollama", version.ref = "testcontainers" }
+testcontainers-kafka = { group="org.testcontainers", name ="kafka", version.ref = "testcontainers"}
 slf4j-api = { group = "org.slf4j", name = "slf4j-api", version.ref = "slf4j" }
 logback-classic = { group = "ch.qos.logback", name = "logback-classic", version.ref = "logback" }
 spring-test = { group = "org.springframework", name = "spring-test" }