[Flaky test] Fix flaky test PulsarFunctionsTest.testExclamationFunction
Fixes
@@ -24,7 +24,6 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import com.google.gson.Gson;
-
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
@@ -35,7 +34,6 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
-
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import net.jodah.failsafe.Failsafe;
@@ -93,6 +91,7 @@ import org.apache.pulsar.tests.integration.io.SourceTester;
import org.apache.pulsar.tests.integration.topologies.FunctionRuntimeType;
import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
import org.assertj.core.api.Assertions;
+import org.awaitility.Awaitility;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.shaded.com.google.common.collect.Sets;
import org.testng.annotations.Test;
@@ -1647,8 +1646,13 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
// update parallelism
updateFunctionParallelism(functionName, 2);
+ Awaitility.await()
+ .pollInterval(Duration.ofMillis(500L))
+ .ignoreExceptions()
+ .untilAsserted(() ->
//get function status
- getFunctionStatus(functionName, 0, true, 2);
+ getFunctionStatus(functionName, 0, true, 2)
+ );
// delete function
deleteFunction(functionName);
