Support Confluent Platform 6 in Kafka (#3293)

* Support Confluent Platform 6 in Kafka

* remove `KafkaContainer#startZookeeper`

Author: bsideup

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

@@ -1,14 +1,12 @@
 package org.testcontainers.containers;
 
-import com.github.dockerjava.api.command.ExecCreateCmdResponse;
 import com.github.dockerjava.api.command.InspectContainerResponse;
 import lombok.SneakyThrows;
 import org.testcontainers.images.builder.Transferable;
 import org.testcontainers.utility.DockerImageName;
 import org.testcontainers.utility.TestcontainersConfiguration;
 
 import java.nio.charset.StandardCharsets;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -108,14 +106,18 @@ protected void containerIsStarting(InspectContainerResponse containerInfo, boole
             return;
         }
 
+        String command = "#!/bin/bash\n";
         final String zookeeperConnect;
         if (externalZookeeperConnect != null) {
             zookeeperConnect = externalZookeeperConnect;
         } else {
-            zookeeperConnect = startZookeeper();
+            zookeeperConnect = "localhost:" + ZOOKEEPER_PORT;
+            command += "echo 'clientPort=" + ZOOKEEPER_PORT + "' > zookeeper.properties\n";
+            command += "echo 'dataDir=/var/lib/zookeeper/data' >> zookeeper.properties\n";
+            command += "echo 'dataLogDir=/var/lib/zookeeper/log' >> zookeeper.properties\n";
+            command += "zookeeper-server-start zookeeper.properties &\n";
         }
 
-        String command = "#!/bin/bash \n";
         command += "export KAFKA_ZOOKEEPER_CONNECT='" + zookeeperConnect + "'\n";
         command += "export KAFKA_ADVERTISED_LISTENERS='" + Stream
             .concat(
@@ -130,22 +132,8 @@ protected void containerIsStarting(InspectContainerResponse containerInfo, boole
         command += "/etc/confluent/docker/launch \n";
 
         copyFileToContainer(
-            Transferable.of(command.getBytes(StandardCharsets.UTF_8), 700),
+            Transferable.of(command.getBytes(StandardCharsets.UTF_8), 0777),
             STARTER_SCRIPT
         );
     }
-
-    @SneakyThrows(InterruptedException.class)
-    private String startZookeeper() {
-        ExecCreateCmdResponse execCreateCmdResponse = dockerClient.execCreateCmd(getContainerId())
-            .withCmd("sh", "-c", "" +
-                "printf 'clientPort=" + ZOOKEEPER_PORT + "\ndataDir=/var/lib/zookeeper/data\ndataLogDir=/var/lib/zookeeper/log' > /zookeeper.properties\n" +
-                "zookeeper-server-start /zookeeper.properties\n"
-            )
-            .exec();
-
-        dockerClient.execStartCmd(execCreateCmdResponse.getId()).start().awaitStarted(10, TimeUnit.SECONDS);
-
-        return "localhost:" + ZOOKEEPER_PORT;
-    }
 }

modules/kafka/src/test/java/org/testcontainers/containers/KafkaContainerTest.java

@@ -99,6 +99,16 @@ public void testExternalZookeeperWithExternalNetwork() throws Exception {
         }
     }
 
+    @Test
+    public void testConfluentPlatformVersion6() throws Exception {
+        try (
+            KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.0.0"))
+        ) {
+            kafka.start();
+            testKafkaFunctionality(kafka.getBootstrapServers());
+        }
+    }
+
     protected void testKafkaFunctionality(String bootstrapServers) throws Exception {
         try (
             KafkaProducer<String, String> producer = new KafkaProducer<>(