Rework KafkaContainer (do not run a sub-process) (#2078)

* Rework KafkaContainer (do not run a sub-process)

* revert unrelated change

* Escape environment variables

* Start ZK before Kafka

Author: bsideup

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

@@ -4,16 +4,22 @@
 import com.github.dockerjava.api.command.InspectContainerResponse;
 import com.github.dockerjava.core.command.ExecStartResultCallback;
 import lombok.SneakyThrows;
+import org.testcontainers.images.builder.Transferable;
 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;
 
 /**
  * This container wraps Confluent Kafka and Zookeeper (optionally)
  *
  */
 public class KafkaContainer extends GenericContainer<KafkaContainer> {
 
+    private static final String STARTER_SCRIPT = "/testcontainers_start.sh";
+
     public static final int KAFKA_PORT = 9093;
 
     public static final int ZOOKEEPER_PORT = 2181;
@@ -97,7 +103,7 @@ public String getBootstrapServers() {
 
     @Override
     protected void doStart() {
-        withCommand("sleep infinity");
+        withCommand("sh", "-c", "while [ ! -f " + STARTER_SCRIPT + " ]; do sleep 0.1; done; " + STARTER_SCRIPT);
 
         if (externalZookeeperConnect == null) {
             addExposedPort(ZOOKEEPER_PORT);
@@ -124,17 +130,24 @@ protected void containerIsStarting(InspectContainerResponse containerInfo, boole
             zookeeperConnect = startZookeeper();
         }
 
-        String internalIp = containerInfo.getNetworkSettings().getIpAddress();
-
-        ExecCreateCmdResponse execCreateCmdResponse = dockerClient.execCreateCmd(getContainerId())
-            .withCmd("sh", "-c", "" +
-                "export KAFKA_ZOOKEEPER_CONNECT=" + zookeeperConnect + "\n" +
-                "export KAFKA_ADVERTISED_LISTENERS=" + getBootstrapServers() + "," + String.format("BROKER://%s:9092", internalIp) + "\n" +
-                "/etc/confluent/docker/run"
+        String command = "#!/bin/bash \n";
+        command += "export KAFKA_ZOOKEEPER_CONNECT='" + zookeeperConnect + "'\n";
+        command += "export KAFKA_ADVERTISED_LISTENERS='" + Stream
+            .concat(
+                Stream.of(getBootstrapServers()),
+                containerInfo.getNetworkSettings().getNetworks().values().stream()
+                    .map(it -> "BROKER://" + it.getIpAddress() + ":9092")
             )
-            .exec();
+            .collect(Collectors.joining(",")) + "'\n";
 
-        dockerClient.execStartCmd(execCreateCmdResponse.getId()).exec(new ExecStartResultCallback()).awaitStarted(10, TimeUnit.SECONDS);
+        command += ". /etc/confluent/docker/bash-config \n";
+        command += "/etc/confluent/docker/configure \n";
+        command += "/etc/confluent/docker/launch \n";
+
+        copyFileToContainer(
+            Transferable.of(command.getBytes(StandardCharsets.UTF_8), 700),
+            STARTER_SCRIPT
+        );
     }
 
     @SneakyThrows(InterruptedException.class)

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

@@ -16,7 +16,6 @@
 import java.util.Arrays;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.tuple;
@@ -46,7 +45,8 @@ public void testExternalZookeeperWithKafkaNetwork() throws Exception {
                 .withNetworkAliases("zookeeper")
                 .withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
         ) {
-            Stream.of(kafka, zookeeper).parallel().forEach(GenericContainer::start);
+            zookeeper.start();
+            kafka.start();
 
             testKafkaFunctionality(kafka.getBootstrapServers());
         }
@@ -66,7 +66,8 @@ public void testExternalZookeeperWithExternalNetwork() throws Exception {
                 .withNetworkAliases("zookeeper")
                 .withEnv("ZOOKEEPER_CLIENT_PORT", "2181");
         ) {
-            Stream.of(kafka, zookeeper).parallel().forEach(GenericContainer::start);
+            zookeeper.start();
+            kafka.start();
 
             testKafkaFunctionality(kafka.getBootstrapServers());
         }