[fix][broker][branch-3.3] Fix namespace unload might be blocked too long with extensible load manager (#23433)

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/BrokersBase.java

@@ -387,7 +387,11 @@ public void healthCheck(@Suspended AsyncResponse asyncResponse,
                     asyncResponse.resume(Response.ok("ok").build());
                 }).exceptionally(ex -> {
                     if (!isRedirectException(ex)) {
-                        LOG.error("[{}] Fail to run health check.", clientAppId(), ex);
+                        if (isNotFoundException(ex)) {
+                            LOG.warn("[{}] Failed to run health check: {}", clientAppId(), ex.getMessage());
+                        } else {
+                            LOG.error("[{}] Failed to run health check.", clientAppId(), ex);
+                        }
                     }
                     resumeAsyncResponseExceptionally(asyncResponse, ex);
                     return null;

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

@@ -88,6 +88,7 @@
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.service.BrokerServiceException;
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.CompressionType;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
@@ -1296,6 +1297,14 @@ private void handleBrokerCreationEvent(String broker) {
                                         broker, cleanupJobs.size());
                             }
                         }
+                    })
+                    .exceptionally(e -> {
+                        if (FutureUtil.unwrapCompletionException(e) instanceof PulsarAdminException.NotFoundException) {
+                            log.warn("{} Failed to run health check: {}", broker, e.getMessage());
+                        } else {
+                            log.error("{} Failed to run health check", broker, e);
+                        }
+                        return null;
                     });
         }
     }
@@ -1328,12 +1337,19 @@ private void handleBrokerDeletionEvent(String broker) {
         }
     }
 
+    private boolean channelDisabled() {
+        final var channelState = this.channelState;
+        if (channelState == Disabled || channelState == Closed) {
+            log.warn("[{}] Skip scheduleCleanup because the state is {} now", brokerId, channelState);
+            return true;
+        }
+        return false;
+    }
+
     private void scheduleCleanup(String broker, long delayInSecs) {
         var scheduled = new MutableObject<CompletableFuture<Void>>();
         try {
-            final var channelState = this.channelState;
-            if (channelState == Disabled || channelState == Closed) {
-                log.warn("[{}] Skip scheduleCleanup because the state is {} now", brokerId, channelState);
+            if (channelDisabled()) {
                 return;
             }
             cleanupJobs.computeIfAbsent(broker, k -> {
@@ -1454,6 +1470,10 @@ private CompletableFuture<Void> healthCheckBrokerAsync(String brokerId) {
     }
 
     private void doHealthCheckBrokerAsyncWithRetries(String brokerId, int retry, CompletableFuture<Void> future) {
+        if (channelDisabled()) {
+            future.complete(null);
+            return;
+        }
         try {
             var admin = getPulsarAdmin();
             admin.brokers().healthcheckAsync(TopicVersion.V2, Optional.of(brokerId))
@@ -1464,7 +1484,6 @@ private void doHealthCheckBrokerAsyncWithRetries(String brokerId, int retry, Com
                             return;
                         }
                         if (retry == MAX_BROKER_HEALTH_CHECK_RETRY) {
-                            log.error("Failed health-check broker :{}", brokerId, e);
                             future.completeExceptionally(FutureUtil.unwrapCompletionException(e));
                         } else {
                             pulsar.getExecutor()
@@ -1501,7 +1520,12 @@ private synchronized void doCleanup(String broker, boolean gracefully) {
                 return;
             } catch (Exception e) {
                 if (debug()) {
-                    log.info("Failed to check broker:{} health", broker, e);
+                    if (e instanceof ExecutionException
+                            && e.getCause() instanceof PulsarAdminException.NotFoundException) {
+                        log.info("The broker {} is not healthy because it's not found", broker);
+                    } else {
+                        log.info("Failed to check broker:{} health", broker, e);
+                    }
                 }
                 log.info("Checked the broker:{} health. Continue the orphan bundle cleanup", broker);
             }

pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerCloseTest.java

@@ -88,14 +88,18 @@ private ServiceConfiguration brokerConfig() {
         config.setLoadManagerClassName(ExtensibleLoadManagerImpl.class.getName());
         config.setLoadBalancerDebugModeEnabled(true);
         config.setBrokerShutdownTimeoutMs(100);
+
+        // Reduce these timeout configs to avoid failed tests being blocked too long
+        config.setMetadataStoreOperationTimeoutSeconds(5);
+        config.setNamespaceBundleUnloadingTimeoutMs(5000);
         return config;
     }
 
 
-    @Test
+    @Test(invocationCount = 10)
     public void testCloseAfterLoadingBundles() throws Exception {
         setupBrokers(3);
-        final var topic = "test";
+        final var topic = "test-" + System.currentTimeMillis();
         final var admin = brokers.get(0).getAdminClient();
         admin.topics().createPartitionedTopic(topic, 20);
         admin.lookups().lookupPartitionedTopic(topic);