Add numFunctionWorkers and externalServices to cluster spec (#2160)

*Motivation*

ClusterSpec is used for defining how a cluster looks like for integration
testing. Add `numFunctionWorker` and `externalServices` in cluster spec,
so we have a common place for setting up a cluster.

*Changes*

- `numFunctionWorkers`: define how many function workers to run in the cluster. When a cluster is created from the cluster spec, it will start
the same number of worker containers.

- `functionRuntimeType`: define how the worker will invoke functions,
whether it is in process mode or thread mode.

- `externalServices`: define whether there are more external services to
run along with the cluster. for example, we need cassandra or kafka for
testing connectors, and we need s3 mock for testing offloaders.

tests/integration/semantics/src/test/java/org/apache/pulsar/tests/integration/functions/runtime/PulsarFunctionsThreadRuntimeTest.java → tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/containers/CassandraContainer.java

@@ -16,28 +16,33 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.tests.integration.functions.runtime;
+package org.apache.pulsar.tests.containers;
 
 import lombok.extern.slf4j.Slf4j;
-import org.testcontainers.containers.Container.ExecResult;
-import org.testng.annotations.BeforeClass;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
 
 /**
- * Run the runtime test cases in thread mode.
+ * Cassandra Container.
  */
 @Slf4j
-public class PulsarFunctionsThreadRuntimeTest extends PulsarFunctionsRuntimeTest {
+public class CassandraContainer<SelfT extends ChaosContainer<SelfT>> extends ChaosContainer<SelfT> {
 
-    public PulsarFunctionsThreadRuntimeTest() {
-        super(RuntimeFactory.THREAD);
-    }
+    public static final String NAME = "cassandra";
+    public static final int PORT = 9042;
 
-    @BeforeClass
-    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());
+    public CassandraContainer(String clusterName) {
+        super(clusterName, "cassandra:3");
     }
 
+    @Override
+    protected void configure() {
+        super.configure();
+        this.withNetworkAliases(NAME)
+            .withExposedPorts(PORT)
+            .withCreateContainerCmdModifier(createContainerCmd -> {
+                createContainerCmd.withHostName(NAME);
+                createContainerCmd.withName(clusterName + "-" + NAME);
+            })
+            .waitingFor(new HostPortWaitStrategy());
+    }
 }

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

@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.containers;
+
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.BindMode;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+
+/**
+ * Cassandra Container.
+ */
+@Slf4j
+public class KafkaContainer<SelfT extends ChaosContainer<SelfT>> extends ChaosContainer<SelfT> {
+
+    public static final String NAME = "kafka";
+    public static final int INTERNAL_PORT = 9092;
+    public static final int PORT = 9093;
+
+    public KafkaContainer(String clusterName) {
+        super(clusterName, "confluentinc/cp-kafka:4.1.1");
+    }
+
+    @Override
+    protected void configure() {
+        super.configure();
+        this.withNetworkAliases(NAME)
+            .withExposedPorts(INTERNAL_PORT, PORT)
+            .withClasspathResourceMapping(
+                "kafka-zookeeper.properties", "/zookeeper.properties",
+                BindMode.READ_ONLY)
+            .withCommand("sh", "-c", "zookeeper-server-start /zookeeper.properties & /etc/confluent/docker/run")
+            .withEnv("KAFKA_LISTENERS",
+                "INTERNAL://kafka:" + INTERNAL_PORT + ",PLAINTEXT://" + "0.0.0.0" + ":" + PORT)
+            .withEnv("KAFKA_ZOOKEEPER_CONNECT", "localhost:2181")
+            .withEnv("KAFKA_LISTENER_SECURITY_PROTOCOL_MAP", "INTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT")
+            .withEnv("KAFKA_INTER_BROKER_LISTENER_NAME", "INTERNAL")
+            .withEnv("KAFKA_BROKER_ID", "1")
+            .withEnv("KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR", "1")
+            .withEnv("KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS", "1")
+            .withEnv("KAFKA_LOG_FLUSH_INTERVAL_MESSAGES", Long.MAX_VALUE + "")
+            .withCreateContainerCmdModifier(createContainerCmd -> {
+                createContainerCmd.withHostName(NAME);
+                createContainerCmd.withName(clusterName + "-" + NAME);
+            })
+            .waitingFor(new HostPortWaitStrategy());
+    }
+}

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

@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.containers;
+
+import lombok.extern.slf4j.Slf4j;
+import org.testcontainers.containers.SocatContainer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+
+/**
+ * Cassandra Container.
+ */
+@Slf4j
+public class KafkaProxyContainer extends SocatContainer {
+
+    public static final String NAME = "kafka-proxy";
+
+    private final String clusterName;
+
+    public KafkaProxyContainer(String clusterName) {
+        super();
+        this.clusterName = clusterName;
+    }
+
+    @Override
+    protected void configure() {
+        super.configure();
+        this.withNetworkAliases(NAME)
+            .withTarget(KafkaContainer.PORT, KafkaContainer.NAME)
+            .withCreateContainerCmdModifier(createContainerCmd -> {
+                createContainerCmd.withHostName(NAME);
+                createContainerCmd.withName(clusterName + "-" + NAME);
+            })
+            .waitingFor(new HostPortWaitStrategy());
+    }
+}

tests/integration/semantics/src/test/java/org/apache/pulsar/tests/integration/functions/runtime/PulsarFunctionsProcessRuntimeTest.java → tests/integration-tests-topologies/src/main/java/org/apache/pulsar/tests/topologies/FunctionRuntimeType.java

@@ -16,26 +16,12 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.tests.integration.functions.runtime;
-
-import lombok.extern.slf4j.Slf4j;
-import org.testcontainers.containers.Container;
-import org.testng.annotations.BeforeClass;
+package org.apache.pulsar.tests.topologies;
 
 /**
- * Run runtime tests in process mode.
+ * Runtime type to run functions.
  */
-@Slf4j
-public class PulsarFunctionsProcessRuntimeTest extends PulsarFunctionsRuntimeTest {
-    public PulsarFunctionsProcessRuntimeTest() {
-        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());
-    }
+public enum FunctionRuntimeType {
+    PROCESS,
+    THREAD
 }

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

@@ -21,20 +21,15 @@
 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;
@@ -49,7 +44,6 @@
 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.
@@ -167,6 +161,27 @@ public void start() throws Exception {
         log.info("Pulsar cluster {} is up running:", clusterName);
         log.info("\tBinary Service Url : {}", getPlainTextServiceUrl());
         log.info("\tHttp Service Url : {}", getHttpServiceUrl());
+
+        // start function workers
+        if (spec.numFunctionWorkers() > 0) {
+            switch (spec.functionRuntimeType()) {
+                case THREAD:
+                    startFunctionWorkersWithThreadContainerFactory(spec.numFunctionWorkers());
+                    break;
+                case PROCESS:
+                    startFunctionWorkersWithProcessContainerFactory(spec.numFunctionWorkers());
+                    break;
+            }
+        }
+
+        // start external services
+        Map<String, GenericContainer<?>> externalServices = spec.externalServices;
+        if (null != externalServices) {
+            externalServices.entrySet().forEach(service -> {
+                service.getValue().start();
+                log.info("Successfully start external service {}.", service.getKey());
+            });
+        }
     }
 
     private static <T extends PulsarContainer> Map<String, T> runNumContainers(String serviceName,
@@ -186,13 +201,15 @@ private static <T extends PulsarContainer> Map<String, T> runNumContainers(Strin
     }
 
     public void 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);
+        Stream.of(proxyContainer, csContainer, zkContainer).parallel().forEach(GenericContainer::stop);
+        workerContainers.values().parallelStream().forEach(GenericContainer::stop);
+        brokerContainers.values().parallelStream().forEach(GenericContainer::stop);
+        bookieContainers.values().parallelStream().forEach(GenericContainer::stop);
+        if (null != spec.externalServices()) {
+            spec.externalServices().values()
+                .parallelStream()
+                .forEach(GenericContainer::stop);
+        }
 
         try {
             network.close();
@@ -201,7 +218,7 @@ public void stop() {
         }
     }
 
-    public void startFunctionWorkersWithProcessContainerFactory(int numFunctionWorkers) {
+    private void startFunctionWorkersWithProcessContainerFactory(int numFunctionWorkers) {
         String serviceUrl = "pulsar://pulsar-broker-0:" + PulsarContainer.BROKER_PORT;
         String httpServiceUrl = "http://pulsar-broker-0:" + PulsarContainer.BROKER_HTTP_PORT;
         workerContainers.putAll(runNumContainers(
@@ -225,7 +242,7 @@ public void startFunctionWorkersWithProcessContainerFactory(int numFunctionWorke
         ));
     }
 
-    public void startFunctionWorkersWithThreadContainerFactory(int numFunctionWorkers) {
+    private void startFunctionWorkersWithThreadContainerFactory(int numFunctionWorkers) {
         String serviceUrl = "pulsar://pulsar-broker-0:" + PulsarContainer.BROKER_PORT;
         String httpServiceUrl = "http://pulsar-broker-0:" + PulsarContainer.BROKER_HTTP_PORT;
         workerContainers.putAll(runNumContainers(

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

@@ -18,11 +18,15 @@
  */
 package org.apache.pulsar.tests.topologies;
 
+import java.util.Map;
 import lombok.Builder;
 import lombok.Builder.Default;
 import lombok.Getter;
 import lombok.Setter;
 import lombok.experimental.Accessors;
+import org.apache.pulsar.tests.containers.ChaosContainer;
+import org.testcontainers.containers.GenericContainer;
+import org.testng.collections.Maps;
 
 /**
  * Spec to build a pulsar cluster.
@@ -64,6 +68,30 @@ public class PulsarClusterSpec {
     @Default
     int numProxies = 1;
 
+    /**
+     * Returns number of function workers.
+     *
+     * @return number of function workers.
+     */
+    @Default
+    int numFunctionWorkers = 0;
+
+    /**
+     * Returns the function runtime type.
+     *
+     * @return the function runtime type.
+     */
+    @Default
+    FunctionRuntimeType functionRuntimeType = FunctionRuntimeType.PROCESS;
+
+    /**
+     * Returns the list of external services to start with
+     * this cluster.
+     *
+     * @return the list of external services to start with the cluster.
+     */
+    Map<String, GenericContainer<?>> externalServices = Maps.newHashMap();
+
     /**
      * Returns the flag whether to enable/disable container log.
      *

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

@@ -62,7 +62,6 @@ public static Object[][] serviceUrls() {
     @BeforeClass
     public void setupCluster() throws Exception {
         this.setupCluster("");
-        pulsarCluster.startFunctionWorkersWithProcessContainerFactory(1);
     }
 
     public void setupCluster(String namePrefix) throws Exception {
@@ -75,7 +74,7 @@ public void setupCluster(String namePrefix) throws Exception {
         setupCluster(spec);
     }
 
-    private void setupCluster(PulsarClusterSpec spec) throws Exception {
+    protected void setupCluster(PulsarClusterSpec spec) throws Exception {
         log.info("Setting up cluster {} with {} bookies, {} brokers",
             spec.clusterName(), spec.numBookies(), spec.numBrokers());
 

tests/integration-tests-topologies/src/main/resources/kafka-zookeeper.properties

@@ -0,0 +1,36 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+clientPort=2181
+dataDir=/var/lib/zookeeper/data
+dataLogDir=/var/lib/zookeeper/log