From b08a23291b8686071b049a8b26eb7dc6014d9bb4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Edd=C3=BA=20Mel=C3=A9ndez=20Gonzales?=
 <eddu.melendez@gmail.com>
Date: Thu, 25 Jul 2024 08:37:23 -0600
Subject: [PATCH] Add support for Apache Kafka native image (#8993)

---
 .../org/testcontainers/kafka/KafkaContainer.java    |  6 ++++--
 .../testcontainers/kafka/KafkaContainerTest.java    | 13 ++++++++++++-
 2 files changed, 16 insertions(+), 3 deletions(-)

diff --git a/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java b/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java
index 86759979cf7..7eb8fa18c8b 100644
--- a/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java
+++ b/modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java
@@ -12,7 +12,7 @@
 /**
  * Testcontainers implementation for Apache Kafka.
  * <p>
- * Supported image: {@code apache/kafka}
+ * Supported image: {@code apache/kafka}, {@code apache/kafka-native}
  * <p>
  * Exposed ports: 9092
  */
@@ -20,6 +20,8 @@ public class KafkaContainer extends GenericContainer<KafkaContainer> {
 
     private static final DockerImageName DEFAULT_IMAGE_NAME = DockerImageName.parse("apache/kafka");
 
+    private static final DockerImageName APACHE_KAFKA_NATIVE_IMAGE_NAME = DockerImageName.parse("apache/kafka-native");
+
     private static final int KAFKA_PORT = 9092;
 
     private static final String DEFAULT_INTERNAL_TOPIC_RF = "1";
@@ -34,7 +36,7 @@ public KafkaContainer(String imageName) {
 
     public KafkaContainer(DockerImageName dockerImageName) {
         super(dockerImageName);
-        dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME);
+        dockerImageName.assertCompatibleWith(DEFAULT_IMAGE_NAME, APACHE_KAFKA_NATIVE_IMAGE_NAME);
 
         withExposedPorts(KAFKA_PORT);
         withEnv("CLUSTER_ID", DEFAULT_CLUSTER_ID);
diff --git a/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java b/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java
index d520cf33f7c..93cb4e6138d 100644
--- a/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java
+++ b/modules/kafka/src/test/java/org/testcontainers/kafka/KafkaContainerTest.java
@@ -1,13 +1,24 @@
 package org.testcontainers.kafka;
 
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 import org.testcontainers.AbstractKafka;
 
+@RunWith(Parameterized.class)
 public class KafkaContainerTest extends AbstractKafka {
 
+    @Parameterized.Parameters(name = "{0}")
+    public static String[] params() {
+        return new String[] { "apache/kafka:3.7.0", "apache/kafka-native:3.8.0-rc3" };
+    }
+
+    @Parameterized.Parameter
+    public String imageName;
+
     @Test
     public void testUsage() throws Exception {
-        try (KafkaContainer kafka = new KafkaContainer("apache/kafka:3.7.0")) {
+        try (KafkaContainer kafka = new KafkaContainer(imageName)) {
             kafka.start();
             testKafkaFunctionality(kafka.getBootstrapServers());
         }