[Tests] Reduce functions integration test flakiness (apache#10074)

* Use unique function names * Use Awaitility for all function status checks - wait up to 15 seconds
Jhanay · Mar 30, 2021 · 4c204c2 · 4c204c2
1 parent 323207c
commit 4c204c2
Showing 1 changed file with 14 additions and 9 deletions.
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
@@ -895,7 +895,7 @@ public void testFunctionLocalRun(Runtime runtime) throws  Exception {
         commandGenerator.setAdminUrl("pulsar://pulsar-broker-0:6650");
         commandGenerator.setSourceTopic(inputTopicName);
         commandGenerator.setSinkTopic(outputTopicName);
-        commandGenerator.setFunctionName("localRunTest");
+        commandGenerator.setFunctionName("localRunTest-" + randomName(8));
         commandGenerator.setRuntime(runtime);
         switch (runtime) {
             case JAVA:
@@ -1640,13 +1640,8 @@ private void testExclamationFunction(Runtime runtime,
         // update parallelism
         updateFunctionParallelism(functionName, 2);
 
-        Awaitility.await()
-                .pollInterval(Duration.ofMillis(500L))
-                .ignoreExceptions()
-                .untilAsserted(() ->
-                //get function status
-                getFunctionStatus(functionName, 0, true, 2)
-        );
+        //get function status
+        getFunctionStatus(functionName, 0, true, 2);
 
         // delete function
         deleteFunction(functionName);
@@ -1993,6 +1988,16 @@ private void getFunctionStatus(String functionName, int numMessages, boolean che
 
     private void getFunctionStatus(String functionName, int numMessages, boolean checkRestarts, int parallelism)
         throws Exception {
+        Awaitility.await()
+                .pollInterval(Duration.ofSeconds(1))
+                .atMost(Duration.ofSeconds(15))
+                .ignoreExceptions()
+                .untilAsserted(() ->
+                        doGetFunctionStatus(functionName, numMessages, checkRestarts, parallelism));
+    }
+
+    private void doGetFunctionStatus(String functionName, int numMessages, boolean checkRestarts, int parallelism)
+            throws Exception {
         ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd(
             PulsarCluster.ADMIN_SCRIPT,
             "functions",
@@ -2228,7 +2233,7 @@ public void testAvroSchemaFunction() throws Exception {
         log.info("testAvroSchemaFunction start ...");
         final String inputTopic = "test-avroschema-input-" + randomName(8);
         final String outputTopic = "test-avroschema-output-" + randomName(8);
-        final String functionName = "test-avroschema-fn-202003241756";
+        final String functionName = "test-avroschema-fn-" + randomName(8);
         final int numMessages = 10;
 
         if (pulsarCluster == null) {