Enhance Pulsar cluster with container logs (#2078)

### Motivation
This is continuation of the existing work to move from the arquiilian framework to test containers library , here we focus on more traceability in the pulsar test containers with logging and migrate the smoke test from the old framework.

### Result
Smoke test based on testcontainers should be removing the older arquillian based version.

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/BKContainer.java

@@ -23,6 +23,8 @@
  */
 public class BKContainer extends PulsarContainer<BKContainer> {
 
+    public static final String NAME = "bookie";
+
     public BKContainer(String clusterName, String hostName) {
         super(
             clusterName, hostName, hostName, "bin/run-bookie.sh", BOOKIE_PORT, INVALID_PORT);

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/BrokerContainer.java

@@ -23,6 +23,8 @@
  */
 public class BrokerContainer extends PulsarContainer<BrokerContainer> {
 
+    public static final String NAME = "pulsar-broker";
+
     public BrokerContainer(String clusterName, String hostName) {
         super(
             clusterName, hostName, hostName, "bin/run-broker.sh", BROKER_PORT, INVALID_PORT);

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/ChaosContainer.java

@@ -104,13 +104,13 @@ public void onNext(Frame item) {
     public ExecResult execCmd(String... cmd) throws Exception {
         String cmdString = StringUtils.join(cmd, " ");
 
-        log.info("DOCKER.exec({}:{}): Executing ...", containerId, cmdString);
+        log.info("DOCKER.exec({}:{}): Executing ...", containerName.substring(1), cmdString);
 
         ExecResult result = execInContainer(cmd);
 
-        log.info("Docker.exec({}:{}): Done", containerId, cmdString);
-        log.info("Docker.exec({}:{}): Stdout -\n{}", containerId, cmdString, result.getStdout());
-        log.info("Docker.exec({}:{}): Stderr -\n{}", containerId, cmdString, result.getStderr());
+        log.info("Docker.exec({}:{}): Done", containerName.substring(1), cmdString);
+        log.info("Docker.exec({}:{}): Stdout -\n{}", containerName.substring(1), cmdString, result.getStdout());
+        log.info("Docker.exec({}:{}): Stderr -\n{}", containerName.substring(1), cmdString, result.getStderr());
 
         return result;
     }

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/ProxyContainer.java

@@ -23,6 +23,8 @@
  */
 public class ProxyContainer extends PulsarContainer<ProxyContainer> {
 
+    public static final String NAME = "pulsar-proxy";
+
     public ProxyContainer(String clusterName, String hostName) {
         super(
             clusterName, hostName, hostName, "bin/run-proxy.sh", BROKER_PORT, BROKER_HTTP_PORT);

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/PulsarContainer.java

@@ -24,6 +24,7 @@
 import java.util.Objects;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.tests.DockerUtils;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
 
 /**
@@ -43,20 +44,20 @@ public abstract class PulsarContainer<SelfT extends PulsarContainer<SelfT>> exte
 
     private final String hostname;
     private final String serviceName;
-    private final String serviceEntrypoint;
+    private final String serviceEntryPoint;
     private final int servicePort;
     private final int httpPort;
 
     public PulsarContainer(String clusterName,
                            String hostname,
                            String serviceName,
-                           String serviceEntrypoint,
+                           String serviceEntryPoint,
                            int servicePort,
                            int httpPort) {
         super(clusterName, IMAGE_NAME);
         this.hostname = hostname;
         this.serviceName = serviceName;
-        this.serviceEntrypoint = serviceEntrypoint;
+        this.serviceEntryPoint = serviceEntryPoint;
         this.servicePort = servicePort;
         this.httpPort = httpPort;
     }
@@ -97,9 +98,9 @@ public void start() {
         this.withCreateContainerCmdModifier(createContainerCmd -> {
             createContainerCmd.withHostName(hostname);
             createContainerCmd.withName(getContainerName());
-            createContainerCmd.withEntrypoint(serviceEntrypoint);
+            createContainerCmd.withEntrypoint(serviceEntryPoint);
         });
-
+        
         super.start();
         log.info("Start pulsar service {} at container {}", serviceName, containerName);
     }

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/WorkerContainer.java

@@ -23,6 +23,8 @@
  */
 public class WorkerContainer extends PulsarContainer<WorkerContainer> {
 
+    public static final String NAME = "pulsar-worker";
+
     public WorkerContainer(String clusterName, String hostname) {
         super(
             clusterName, hostname, hostname, "bin/run-functions-worker.sh", -1, BROKER_HTTP_PORT);

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/ZKContainer.java

@@ -23,7 +23,7 @@
 /**
  * A pulsar container that runs zookeeper.
  */
-public class ZKContainer<SelfT extends ZKContainer<SelfT>> extends PulsarContainer<SelfT> {
+public class ZKContainer<SelfT extends PulsarContainer<SelfT>> extends PulsarContainer<SelfT> {
 
     public static final String NAME = "zookeeper";
 
@@ -42,14 +42,14 @@ public ZKContainer(String clusterName) {
     public ZKContainer(String clusterName,
                        String hostname,
                        String serviceName,
-                       String serviceEntrypoint,
+                       String serviceEntryPoint,
                        int servicePort,
                        int httpPort) {
         super(
             clusterName,
             hostname,
             serviceName,
-            serviceEntrypoint,
+            serviceEntryPoint,
             servicePort,
             httpPort);
     }

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/topologies/PulsarCluster.java

@@ -21,14 +21,22 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static org.apache.pulsar.tests.containers.PulsarContainer.CS_PORT;
 
+import com.github.dockerjava.api.command.CreateContainerCmd;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.tests.containers.BKContainer;
@@ -39,7 +47,9 @@
 import org.apache.pulsar.tests.containers.WorkerContainer;
 import org.apache.pulsar.tests.containers.ZKContainer;
 import org.testcontainers.containers.Container.ExecResult;
+import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
 
 /**
  * Pulsar Cluster in containers.
@@ -61,6 +71,7 @@ public static PulsarCluster forSpec(PulsarClusterSpec spec) {
     }
 
     private final PulsarClusterSpec spec;
+
     @Getter
     private final String clusterName;
     private final Network network;
@@ -72,10 +83,13 @@ public static PulsarCluster forSpec(PulsarClusterSpec spec) {
     private final ProxyContainer proxyContainer;
 
     private PulsarCluster(PulsarClusterSpec spec) {
+
         this.spec = spec;
         this.clusterName = spec.clusterName();
         this.network = Network.newNetwork();
-        this.zkContainer = (ZKContainer) new ZKContainer(clusterName)
+
+        this.zkContainer = new ZKContainer(clusterName);
+        this.zkContainer
             .withNetwork(network)
             .withNetworkAliases(ZKContainer.NAME)
             .withEnv("clusterName", clusterName)
@@ -86,10 +100,12 @@ private PulsarCluster(PulsarClusterSpec spec) {
         this.csContainer = new CSContainer(clusterName)
             .withNetwork(network)
             .withNetworkAliases(CSContainer.NAME);
+
         this.bookieContainers = Maps.newTreeMap();
         this.brokerContainers = Maps.newTreeMap();
         this.workerContainers = Maps.newTreeMap();
-        this.proxyContainer = new ProxyContainer(clusterName, "pulsar-proxy")
+
+        this.proxyContainer = new ProxyContainer(clusterName, ProxyContainer.NAME)
             .withNetwork(network)
             .withNetworkAliases("pulsar-proxy")
             .withEnv("zkServers", ZKContainer.NAME)
@@ -170,12 +186,14 @@ private static <T extends PulsarContainer> Map<String, T> runNumContainers(Strin
     }
 
     public void stop() {
-        proxyContainer.stop();
-        workerContainers.values().forEach(WorkerContainer::stop);
-        brokerContainers.values().forEach(BrokerContainer::stop);
-        bookieContainers.values().forEach(BKContainer::stop);
-        csContainer.stop();
-        zkContainer.stop();
+
+        Stream<GenericContainer> list1 = Stream.of(proxyContainer, csContainer, zkContainer);
+        Stream<GenericContainer> list2 =
+            Stream.of(workerContainers.values(), brokerContainers.values(), bookieContainers.values())
+                .flatMap(Collection::stream);
+        Stream<GenericContainer> list3 = Stream.concat(list1, list2);
+        list3.parallel().forEach(GenericContainer::stop);
+
         try {
             network.close();
         } catch (Exception e) {

tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/topologies/PulsarClusterTestBase.java

@@ -18,14 +18,18 @@
  */
 package org.apache.pulsar.tests.topologies;
 
-import java.util.concurrent.ThreadLocalRandom;
 import lombok.extern.slf4j.Slf4j;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.DataProvider;
 
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.stream.Stream;
+
+import static java.util.stream.Collectors.joining;
+
 @Slf4j
-public class PulsarClusterTestBase {
+public abstract class PulsarClusterTestBase {
 
     @DataProvider(name = "ServiceUrlAndTopics")
     public static Object[][] serviceUrlAndTopics() {
@@ -56,19 +60,22 @@ public static Object[][] serviceUrls() {
     protected static PulsarCluster pulsarCluster;
 
     @BeforeClass
-    public static void setupCluster() throws Exception {
-        StringBuilder sb = new StringBuilder();
-        for (int i = 0; i < 8; i++) {
-            sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a'));
-        }
+    public void setupCluster() throws Exception {
+        this.setupCluster("");
+        pulsarCluster.startFunctionWorkersWithProcessContainerFactory(1);
+    }
+
+    public void setupCluster(String namePrefix) throws Exception {
         PulsarClusterSpec spec = PulsarClusterSpec.builder()
-            .clusterName(sb.toString())
+            .clusterName(Stream.of(this.getClass().getSimpleName(), namePrefix, randomName(5))
+                .filter(s -> s != null && !s.isEmpty())
+                .collect(joining("-")))
             .build();
 
         setupCluster(spec);
     }
 
-    protected static void setupCluster(PulsarClusterSpec spec) throws Exception {
+    private void setupCluster(PulsarClusterSpec spec) throws Exception {
         log.info("Setting up cluster {} with {} bookies, {} brokers",
             spec.clusterName(), spec.numBookies(), spec.numBrokers());
 
@@ -79,14 +86,14 @@ protected static void setupCluster(PulsarClusterSpec spec) throws Exception {
     }
 
     @AfterClass
-    public static void teardownCluster() {
+    public void tearDownCluster() {
         if (null != pulsarCluster) {
             pulsarCluster.stop();
         }
     }
 
     protected static String randomName(int numChars) {
-        StringBuilder sb = new StringBuilder();;;;
+        StringBuilder sb = new StringBuilder();
         for (int i = 0; i < 8; i++) {
             sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a'));
         }

tests/integration/semantics/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.tests.integration.functions;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.tests.integration.functions.runtime.PulsarFunctionsRuntimeTest;
 import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator.Runtime;
 import org.apache.pulsar.tests.topologies.PulsarClusterTestBase;
 import org.testcontainers.containers.Container.ExecResult;
@@ -49,16 +50,6 @@ protected static String getExclamationClass(Runtime runtime) {
         }
     }
 
-    @BeforeClass
-    public static void setupCluster() throws Exception {
-        PulsarClusterTestBase.setupCluster();
-
-        pulsarCluster.startFunctionWorkersWithProcessContainerFactory(1);
-
-        ExecResult result = pulsarCluster.getAnyWorker().execCmd("cat", "/pulsar/conf/functions_worker.yml");
-        log.info("Functions Worker Config : \n{}", result.getStdout());
-    }
-
     @DataProvider(name = "FunctionRuntimes")
     public static Object[][] functionRuntimes() {
         return new Object[][] {

tests/integration/semantics/src/test/java/org/apache/pulsar/tests/integration/functions/runtime/PulsarFunctionsProcessRuntimeTest.java

@@ -18,11 +18,24 @@
  */
 package org.apache.pulsar.tests.integration.functions.runtime;
 
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.Container;
+import org.testng.annotations.BeforeClass;
+
 /**
  * Run runtime tests in process mode.
  */
+@Slf4j
 public class PulsarFunctionsProcessRuntimeTest extends PulsarFunctionsRuntimeTest {
     public PulsarFunctionsProcessRuntimeTest() {
-        super(ContainerFactory.PROCESS);
+        super(RuntimeFactory.PROCESS);
+    }
+
+    @BeforeClass
+    public void setupCluster() throws Exception {
+        super.setupCluster(RuntimeFactory.PROCESS.toString());
+        pulsarCluster.startFunctionWorkersWithProcessContainerFactory(1);
+        Container.ExecResult result = pulsarCluster.getAnyWorker().execCmd("cat", "/pulsar/conf/functions_worker.yml");
+        log.info("Functions Worker Config : \n{}", result.getStdout());
     }
 }

tests/integration/semantics/src/test/java/org/apache/pulsar/tests/integration/functions/runtime/PulsarFunctionsRuntimeTest.java

@@ -40,15 +40,15 @@
  */
 public abstract class PulsarFunctionsRuntimeTest extends PulsarFunctionsTestBase {
 
-    protected enum ContainerFactory {
+    public enum RuntimeFactory {
         PROCESS,
         THREAD
     }
 
-    private final ContainerFactory containerFactory;
+    private final RuntimeFactory runtimeFactory;
 
-    public PulsarFunctionsRuntimeTest(ContainerFactory containerFactory) {
-        this.containerFactory = containerFactory;
+    public PulsarFunctionsRuntimeTest(RuntimeFactory runtimeFactory) {
+        this.runtimeFactory = runtimeFactory;
     }
 
     //
@@ -57,7 +57,7 @@ public PulsarFunctionsRuntimeTest(ContainerFactory containerFactory) {
 
     @Test(dataProvider = "FunctionRuntimes")
     public void testExclamationFunction(Runtime runtime) throws Exception {
-        if (ContainerFactory.THREAD == containerFactory && Runtime.PYTHON == runtime) {
+        if (runtimeFactory == RuntimeFactory.THREAD && runtime == Runtime.PYTHON) {
             // python can only run on process mode
             return;
         }

tests/integration/semantics/src/test/java/org/apache/pulsar/tests/integration/functions/runtime/PulsarFunctionsThreadRuntimeTest.java

@@ -19,7 +19,6 @@
 package org.apache.pulsar.tests.integration.functions.runtime;
 
 import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.tests.topologies.PulsarClusterTestBase;
 import org.testcontainers.containers.Container.ExecResult;
 import org.testng.annotations.BeforeClass;
 
@@ -30,15 +29,13 @@
 public class PulsarFunctionsThreadRuntimeTest extends PulsarFunctionsRuntimeTest {
 
     public PulsarFunctionsThreadRuntimeTest() {
-        super(ContainerFactory.THREAD);
+        super(RuntimeFactory.THREAD);
     }
 
     @BeforeClass
-    public static void setupCluster() throws Exception {
-        PulsarClusterTestBase.setupCluster();
-
+    public void setupCluster() throws Exception {
+        super.setupCluster(RuntimeFactory.THREAD.toString());
         pulsarCluster.startFunctionWorkersWithThreadContainerFactory(1);
-
         ExecResult result = pulsarCluster.getAnyWorker().execCmd("cat", "/pulsar/conf/functions_worker.yml");
         log.info("Functions Worker Config : \n{}", result.getStdout());
     }