Add support for Apache Kafka native image (#8993)

Author: eddumelendez

modules/kafka/src/main/java/org/testcontainers/kafka/KafkaContainer.java

@@ -12,14 +12,16 @@
 /**
  * Testcontainers implementation for Apache Kafka.
  * <p>
- * Supported image: {@code apache/kafka}
+ * Supported image: {@code apache/kafka}, {@code apache/kafka-native}
  * <p>
  * Exposed ports: 9092
  */
 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);

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());
         }