[Issue-9889] [integration tests] Refactored Function integration tests (

apache#10140) Co-authored-by: David Kjerrumgaard <[email protected]>
lujp · May 11, 2021 · a161e75 · a161e75
1 parent 4d2d66d
commit a161e75
Show file tree

Hide file tree

Showing 43 changed files with 1,893 additions and 1,307 deletions.
diff --git a/.gitignore b/.gitignore
@@ -35,6 +35,7 @@ nb-configuration.xml
 
 # Mac
 **/.DS_Store
+.java-version
 
 # VisualStudioCode artifacts
 .vscode/

diff --git a/...tion/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/...tion/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
diff --git a/.../src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java b/.../src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTestBase.java
@@ -32,54 +32,9 @@
 @Slf4j
 public abstract class PulsarFunctionsTestBase extends PulsarTestSuite {
 
-    @DataProvider(name = "FunctionRuntimeTypes")
-    public static Object[][] getData() {
-        return new Object[][] {
-            { FunctionRuntimeType.PROCESS },
-            { FunctionRuntimeType.THREAD }
-        };
-    }
-
-    protected final FunctionRuntimeType functionRuntimeType;
-
-    public PulsarFunctionsTestBase() {
-        this(FunctionRuntimeType.PROCESS);
-    }
-
-    protected PulsarFunctionsTestBase(FunctionRuntimeType functionRuntimeType) {
-        this.functionRuntimeType = functionRuntimeType;
-    }
-
-    private void setupFunctionWorkers() {
-        final int numFunctionWorkers = 2;
-        log.info("Setting up {} function workers : function runtime type = {}",
-            numFunctionWorkers, functionRuntimeType);
-        pulsarCluster.setupFunctionWorkers(randomName(5), functionRuntimeType, numFunctionWorkers);
-        log.info("{} function workers has started", numFunctionWorkers);
-    }
-
-    private void teardownFunctionWorkers() {
-        log.info("Tearing down function workers ...");
-        pulsarCluster.stopWorkers();
-        log.info("All functions workers are stopped.");
-    }
-
-    @Override
-    public void setupCluster() throws Exception {
-        super.setupCluster();
-        setupFunctionWorkers();
-    }
-
-    @Override
-    public void tearDownCluster() throws Exception {
-        teardownFunctionWorkers();
-        super.tearDownCluster();
-    }
-
     //
     // Common Variables used by functions test
     //
-
     public static final String EXCLAMATION_JAVA_CLASS =
         "org.apache.pulsar.functions.api.examples.ExclamationFunction";
 
@@ -106,7 +61,6 @@ public void tearDownCluster() throws Exception {
 
     public static final String PUBLISH_PYTHON_CLASS = "typed_message_builder_publish.TypedMessageBuilderPublish";
     public static final String EXCEPTION_PYTHON_CLASS = "exception_function";
-
     public static final String EXCLAMATION_PYTHON_FILE = "exclamation_function.py";
     public static final String EXCLAMATION_WITH_DEPS_PYTHON_FILE = "exclamation_with_extra_deps.py";
     public static final String EXCLAMATION_PYTHON_ZIP_FILE = "exclamation.zip";
@@ -119,6 +73,49 @@ public void tearDownCluster() throws Exception {
     public static final String LOGGING_JAVA_CLASS =
             "org.apache.pulsar.functions.api.examples.LoggingFunction";
 
+    @DataProvider(name = "FunctionRuntimeTypes")
+    public static Object[][] getData() {
+        return new Object[][] {
+            { FunctionRuntimeType.PROCESS },
+            { FunctionRuntimeType.THREAD }
+        };
+    }
+
+    @DataProvider(name = "FunctionRuntimes")
+    public static Object[][] functionRuntimes() {
+        return new Object[][] {
+            new Object[] { Runtime.JAVA },
+            new Object[] { Runtime.PYTHON },
+            new Object[] { Runtime.GO }
+        };
+    }
+
+    protected final FunctionRuntimeType functionRuntimeType;
+
+    public PulsarFunctionsTestBase() {
+        this(FunctionRuntimeType.PROCESS);
+    }
+
+    protected PulsarFunctionsTestBase(FunctionRuntimeType functionRuntimeType) {
+        this.functionRuntimeType = functionRuntimeType;
+    }
+
+    @BeforeClass(alwaysRun = true)
+    public void setupFunctionWorkers() {
+        final int numFunctionWorkers = 2;
+        log.info("Setting up {} function workers : function runtime type = {}",
+            numFunctionWorkers, functionRuntimeType);
+        pulsarCluster.setupFunctionWorkers(randomName(5), functionRuntimeType, numFunctionWorkers);
+        log.info("{} function workers has started", numFunctionWorkers);
+    }
+
+    @AfterClass(alwaysRun = true)
+    public void teardownFunctionWorkers() {
+        log.info("Tearing down function workers ...");
+        pulsarCluster.stopWorkers();
+        log.info("All functions workers are stopped.");
+    }
+
     protected static String getExclamationClass(Runtime runtime,
                                                 boolean pyZip,
                                                 boolean extraDeps) {
@@ -136,14 +133,4 @@ protected static String getExclamationClass(Runtime runtime,
             throw new IllegalArgumentException("Unsupported runtime : " + runtime);
         }
     }
-
-    @DataProvider(name = "FunctionRuntimes")
-    public static Object[][] functionRuntimes() {
-        return new Object[][] {
-            new Object[] { Runtime.JAVA },
-            new Object[] { Runtime.PYTHON },
-            new Object[] { Runtime.GO }
-        };
-    }
-
 }
diff --git a/.../functions/PulsarFunctionsThreadTest.java → ...ions/go/PulsarFunctionsGoProcessTest.java b/.../functions/PulsarFunctionsThreadTest.java → ...ions/go/PulsarFunctionsGoProcessTest.java
@@ -16,15 +16,14 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.tests.integration.functions;
+package org.apache.pulsar.tests.integration.functions.go;
 
 import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
 
-/**
- * Thread based test.
- */
-public class PulsarFunctionsThreadTest extends PulsarFunctionsTest {
-    public PulsarFunctionsThreadTest() {
-        super(FunctionRuntimeType.THREAD);
-    }
+public class PulsarFunctionsGoProcessTest extends PulsarFunctionsGoTest {
+
+	PulsarFunctionsGoProcessTest() {
+		super(FunctionRuntimeType.PROCESS);
+	}
+
 }
diff --git a/...src/test/java/org/apache/pulsar/tests/integration/functions/go/PulsarFunctionsGoTest.java b/...src/test/java/org/apache/pulsar/tests/integration/functions/go/PulsarFunctionsGoTest.java
@@ -0,0 +1,37 @@
+/**
+ * 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.integration.functions.go;
+
+import org.apache.pulsar.tests.integration.functions.PulsarFunctionsTest;
+import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator.Runtime;
+import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
+import org.testng.annotations.Test;
+
+public class PulsarFunctionsGoTest extends PulsarFunctionsTest {
+
+	PulsarFunctionsGoTest(FunctionRuntimeType functionRuntimeType) {
+		super(functionRuntimeType);
+	}
+
+    @Test(enabled = false, groups = {"go_function", "function"})
+    public void testGoFunctionLocalRun() throws Exception {
+        testFunctionLocalRun(Runtime.GO);
+    }
+
+}
diff --git a/...functions/PulsarFunctionsProcessTest.java → ...tions/go/PulsarFunctionsGoThreadTest.java b/...functions/PulsarFunctionsProcessTest.java → ...tions/go/PulsarFunctionsGoThreadTest.java
@@ -16,15 +16,13 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.tests.integration.functions;
+package org.apache.pulsar.tests.integration.functions.go;
 
 import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
 
-/**
- * Process based test.
- */
-public class PulsarFunctionsProcessTest extends PulsarFunctionsTest {
-    public PulsarFunctionsProcessTest() {
-        super(FunctionRuntimeType.PROCESS);
-    }
+public class PulsarFunctionsGoThreadTest extends PulsarFunctionsGoTest {
+
+	PulsarFunctionsGoThreadTest() {
+		super(FunctionRuntimeType.THREAD);
+	}
 }
diff --git a/...va/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaProcessTest.java b/...va/org/apache/pulsar/tests/integration/functions/java/PulsarFunctionsJavaProcessTest.java
@@ -0,0 +1,29 @@
+/**
+ * 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.integration.functions.java;
+
+import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
+
+public class PulsarFunctionsJavaProcessTest extends PulsarFunctionsJavaTest {
+
+	public PulsarFunctionsJavaProcessTest() {
+		super(FunctionRuntimeType.PROCESS);
+	}
+
+}