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.
  * 
- * Supported image: {@code apache/kafka}
+ * Supported image: {@code apache/kafka}, {@code apache/kafka-native}
  * 
  * Exposed ports: 9092
  */
@@ -20,6 +20,8 @@ public class KafkaContainer extends GenericContainer {
 
     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());
         }