(pulsar) branch master updated: [fix][test] Fix flaky PulsarFunctionTlsTest.testFunctionsCreation() test (#25889)

[lh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new a6af80198f2 [fix][test] Fix flaky PulsarFunctionTlsTest.testFunctionsCreation() test (#25889)
a6af80198f2 is described below

commit a6af80198f26d31e991e880db3f9b7d601a579c2
Author: Oneby Wang <44...@users.noreply.github.com>
AuthorDate: Sat May 30 16:16:16 2026 +0800

    [fix][test] Fix flaky PulsarFunctionTlsTest.testFunctionsCreation() test (#25889)
---
 .../functions/worker/PulsarFunctionTlsTest.java    | 37 +++++++++++-----------
 1 file changed, 18 insertions(+), 19 deletions(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
index 9727dfb8f41..3d2e5f29886 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/functions/worker/PulsarFunctionTlsTest.java
@@ -19,8 +19,6 @@
 package org.apache.pulsar.functions.worker;
 
 import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertTrue;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.Sets;
@@ -41,10 +39,10 @@ import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Authentication;
 import org.apache.pulsar.client.impl.auth.AuthenticationTls;
 import org.apache.pulsar.common.functions.FunctionConfig;
-import org.apache.pulsar.common.functions.WorkerInfo;
 import org.apache.pulsar.common.policies.data.TenantInfo;
 import org.apache.pulsar.common.util.ClassLoaderUtils;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
@@ -259,23 +257,16 @@ public class PulsarFunctionTlsTest {
 
             log.info().attr("function", functionName).log("Start test function");
 
-            int finalI = i;
-            // Wait for a leader to be ready and create the function.
-            // The createFunctionWithUrl call is included in the retry loop because a leadership
-            // transition can happen between the leader check and the actual API call, causing
-            // a 503 "Leader not yet ready" error.
             final PulsarAdmin createAdmin = pulsarAdmins[i];
-            Awaitility.await().atMost(1, TimeUnit.MINUTES).pollInterval(1, TimeUnit.SECONDS).untilAsserted(() -> {
-                final PulsarWorkerService workerService = ((PulsarWorkerService) fnWorkerServices[finalI]);
-                final LeaderService leaderService = workerService.getLeaderService();
-                assertNotNull(leaderService);
-                if (!leaderService.isLeader()) {
-                    final WorkerInfo workerInfo = workerService.getMembershipManager().getLeader();
-                    assertTrue(workerInfo != null
-                            && !workerInfo.getWorkerId().equals(workerService.getWorkerConfig().getWorkerId()));
-                }
-                createAdmin.functions().createFunctionWithUrl(functionConfig, jarFilePathUrl);
-            });
+            // During function-worker leadership election/switchover, the coordination topic can already point to
+            // the new leader while that worker is still finishing its leader initialization. In that short window
+            // the internal /functions/leader request returns a transient 503 "Leader not yet ready", so retry only
+            // that condition and let all other failures surface immediately.
+            Awaitility.await().atMost(1, TimeUnit.MINUTES)
+                    .pollInterval(1, TimeUnit.SECONDS)
+                    .ignoreExceptionsMatching(PulsarFunctionTlsTest::isLeaderNotReady)
+                    .untilAsserted(() -> createAdmin.functions()
+                            .createFunctionWithUrl(functionConfig, jarFilePathUrl));
 
             // Function creation is not strongly consistent, so this test can fail with a get that is too eager and
             // does not have retries.
@@ -291,6 +282,14 @@ public class PulsarFunctionTlsTest {
             pulsarAdmins[i].functions().deleteFunction(config.getTenant(), config.getNamespace(), config.getName());
         }
     }
+
+    private static boolean isLeaderNotReady(Throwable e) {
+        return e instanceof PulsarAdminException
+                && ((PulsarAdminException) e).getStatusCode() == 503
+                && String.valueOf(((PulsarAdminException) e).getHttpError())
+                        .contains("Leader not yet ready");
+    }
+
     @SuppressWarnings("deprecation")
 
     protected static FunctionConfig createFunctionConfig(