From af3ee892506707e5b956306d7b46f1db2fc0ea49 Mon Sep 17 00:00:00 2001
From: Matteo Merli <mmerli@apache.org>
Date: Mon, 8 Jun 2026 10:02:47 -0700
Subject: [PATCH] [fix][test] Fix flaky ExtensibleLoadManagerImplTest by
 re-serving the channel topic in initializeState

The initializeState @BeforeMethod could time out after 2 minutes when a prior
test (e.g. testHandleNoChannelOwner) churned leader election and left the
channel-topic bundle (pulsar/system, loadbalancer-service-unit-state) in an
owner-recorded-but-unserved state. monitor() does not heal that state
(handleNoChannelOwnerError only restarts election on the "no channel owner now"
error), so the namespace unload could never publish to the channel.

Force-serve the channel topic on every retry attempt via an admin lookup +
getStats (the same sequence awaitChannelOwnerStable already uses), so the unload
can proceed. Guarded to the system-topic table-view variant.
---
 .../ExtensibleLoadManagerImplBaseTest.java       | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java
index 1a6b7e6ddb406..8e4fc7bf4ddbe 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java
@@ -200,16 +200,24 @@ protected void cleanup() throws Exception {
     protected void initializeState() throws PulsarAdminException, IllegalAccessException {
         // After a prior test churned leader election, the channel-topic bundle can be left
         // unserved ("not served by this instance"), making the unload's channel publish fail
-        // (HTTP 500) or hang server-side until the background monitor task (120s interval)
-        // reconciles the brokers' roles with the channel ownership. Drive monitor() eagerly to
-        // heal that state, bound each unload attempt (a synchronous unload() can block longer
-        // than the whole retry window), and fail loudly on exhaustion.
+        // (HTTP 500) or hang server-side. monitor() only self-heals when there is *no* channel
+        // owner; it does NOT heal the case where an owner is recorded but the bundle is not
+        // actually served, so the unload below can never publish. Force-serve the channel topic
+        // each attempt: an admin lookup re-assigns the pulsar/system bundle and getStats makes
+        // the owner load the topic (the lookup layer alone can claim an owner that refuses to
+        // serve). Bound each unload attempt and fail loudly on exhaustion.
+        boolean systemTopicChannel =
+                serviceUnitStateTableViewClassName.equals(ServiceUnitStateTableViewImpl.class.getName());
         Awaitility.await().atMost(120, TimeUnit.SECONDS)
                 .pollInterval(1, TimeUnit.SECONDS)
                 .ignoreExceptions()
                 .untilAsserted(() -> {
                     primaryLoadManager.monitor();
                     secondaryLoadManager.monitor();
+                    if (systemTopicChannel) {
+                        admin.lookups().lookupTopic(ServiceUnitStateTableViewImpl.TOPIC);
+                        admin.topics().getStats(ServiceUnitStateTableViewImpl.TOPIC);
+                    }
                     admin.namespaces().unloadAsync(defaultTestNamespace).get(15, TimeUnit.SECONDS);
                 });
         reset(primaryLoadManager, secondaryLoadManager);