From f40081b21f815175b5d34fa2e89ce7dd7259e3dc Mon Sep 17 00:00:00 2001
From: Vedant Madane <6527493+VedantMadane@users.noreply.github.com>
Date: Sun, 9 Aug 2026 12:30:41 +0000
Subject: [PATCH 1/2] [improve][broker] Avoid blocking joins in
 ClusterResources sync APIs

ClusterResources.isClusterUsed used MetadataCache.getChildren().join()
without a timeout, which can block caller threads (and risk metadata
thread starvation if used from completion paths).

- Implement isClusterUsed via isClusterUsedAsync with operation timeout
- Add getNamespacesForClusterAsync; route sync method through it
- After waitForAll in isClusterUsedAsync, use getNow instead of join

Part of #22544
---
 .../broker/resources/ClusterResources.java    | 41 ++++++++++++++-----
 1 file changed, 31 insertions(+), 10 deletions(-)

diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ClusterResources.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ClusterResources.java
index b0cc50edf1f1d..7e6ebe32ca14e 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ClusterResources.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/ClusterResources.java
@@ -23,6 +23,8 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -71,8 +73,21 @@ public CompletableFuture<Optional<ClusterData>> getClusterAsync(String clusterNa
         return getAsync(joinPath(BASE_CLUSTERS_PATH, clusterName));
     }
 
+    public CompletableFuture<List<String>> getNamespacesForClusterAsync(String tenant, String clusterName) {
+        return getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant, clusterName));
+    }
+
     public List<String> getNamespacesForCluster(String tenant, String clusterName) throws MetadataStoreException {
-        return getChildren(joinPath(BASE_POLICIES_PATH, tenant, clusterName));
+        try {
+            return getNamespacesForClusterAsync(tenant, clusterName)
+                    .get(getOperationTimeoutSec(), TimeUnit.SECONDS);
+        } catch (ExecutionException e) {
+            throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause()
+                    : new MetadataStoreException(e.getCause());
+        } catch (Exception e) {
+            throw new MetadataStoreException(
+                    "Failed to get namespaces for tenant " + tenant + " in cluster " + clusterName, e);
+        }
     }
 
     public void createCluster(String clusterName, ClusterData clusterData) throws MetadataStoreException {
@@ -109,22 +124,28 @@ public CompletableFuture<Boolean> isClusterUsedAsync(String clusterName) {
                             .collect(Collectors.toList());
                     return FutureUtil.waitForAll(futures)
                             .thenApply(__ -> {
-                                // We found a tenant that has at least a namespace in this cluster
-                                return futures.stream().map(CompletableFuture::join)
+                                // Futures are already complete after waitForAll; use getNow to avoid
+                                // accidental blocking if this is ever refactored.
+                                return futures.stream()
+                                        .map(f -> f.getNow(List.of()))
                                         .anyMatch(CollectionUtils::isNotEmpty);
                             });
                 });
     }
 
+    /**
+     * Synchronous wrapper around {@link #isClusterUsedAsync(String)}.
+     * Prefer the async method so callers do not block metadata store threads.
+     */
     public boolean isClusterUsed(String clusterName) throws MetadataStoreException {
-        for (String tenant : getCache().getChildren(BASE_POLICIES_PATH).join()) {
-            if (!getCache().getChildren(joinPath(BASE_POLICIES_PATH, tenant, clusterName)).join().isEmpty()) {
-                // We found a tenant that has at least a namespace in this cluster
-                return true;
-            }
+        try {
+            return isClusterUsedAsync(clusterName).get(getOperationTimeoutSec(), TimeUnit.SECONDS);
+        } catch (ExecutionException e) {
+            throw (e.getCause() instanceof MetadataStoreException) ? (MetadataStoreException) e.getCause()
+                    : new MetadataStoreException(e.getCause());
+        } catch (Exception e) {
+            throw new MetadataStoreException("Failed to check if cluster is used: " + clusterName, e);
         }
-
-        return false;
     }
 
     public boolean clusterExists(String clusterName) throws MetadataStoreException {

From eacf619775306c7dec137491762cb09540b9c3c5 Mon Sep 17 00:00:00 2001
From: Vedant Madane <6527493+VedantMadane@users.noreply.github.com>
Date: Tue, 11 Aug 2026 10:08:34 +0530
Subject: [PATCH 2/2] Trigger CI recalculation