You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/06/22 12:03:45 UTC
(pulsar) 15/17: [fix][broker] Avoid blocking the bundle-throughput lookup on per-bundle metadata reads (#26054)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 9f4d11e63d836b92ee013973d31c7e85c877d19c
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Thu Jun 18 04:25:00 2026 -0700
[fix][broker] Avoid blocking the bundle-throughput lookup on per-bundle metadata reads (#26054)
(cherry picked from commit 561f373c1442a81364154aa9880fcb7ada9b89ef)
---
.../broker/loadbalance/ModularLoadManager.java | 14 ++++
.../loadbalance/impl/ModularLoadManagerImpl.java | 89 ++++++++++++----------
.../common/naming/NamespaceBundleFactory.java | 31 +++++---
.../impl/ModularLoadManagerImplTest.java | 2 +-
.../broker/namespace/NamespaceServiceTest.java | 2 +-
5 files changed, 84 insertions(+), 54 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
index d608bd6784f2..93059d722e73 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManager.java
@@ -138,8 +138,22 @@ public interface ModularLoadManager {
*
* @param bundle
* @return bundle data
+ * @deprecated use {@link #getBundleDataOrDefaultAsync(String)} instead; this method blocks on
+ * metadata-store reads and must not be called from async or event-loop threads.
*/
+ @Deprecated
BundleData getBundleDataOrDefault(String bundle);
+ /**
+ * Asynchronously fetch bundle's load report data.
+ *
+ * @param bundle
+ * @return future of the bundle data
+ */
+ @SuppressWarnings("deprecation")
+ default CompletableFuture<BundleData> getBundleDataOrDefaultAsync(String bundle) {
+ return CompletableFuture.completedFuture(getBundleDataOrDefault(bundle));
+ }
+
String setNamespaceBundleAffinity(String bundle, String broker);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
index a9d7ddd78e07..214d12e7ad2c 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java
@@ -381,46 +381,55 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
// Attempt to local the data for the given bundle in metadata store
// If it cannot be found, return the default bundle data.
+ /**
+ * @deprecated use {@link #getBundleDataOrDefaultAsync(String)} instead.
+ */
+ @Deprecated
@Override
public BundleData getBundleDataOrDefault(final String bundle) {
- BundleData bundleData = null;
- try {
- Optional<BundleData> optBundleData =
- pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle).join();
- if (optBundleData.isPresent()) {
- return optBundleData.get();
- }
+ return getBundleDataOrDefaultAsync(bundle).join();
+ }
- Optional<ResourceQuota> optQuota = pulsarResources.getLoadBalanceResources().getQuotaResources()
- .getQuota(bundle).join();
- if (optQuota.isPresent()) {
- ResourceQuota quota = optQuota.get();
- bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES);
- // Initialize from existing resource quotas if new API ZNodes do not exist.
- final TimeAverageMessageData shortTermData = bundleData.getShortTermData();
- final TimeAverageMessageData longTermData = bundleData.getLongTermData();
-
- shortTermData.setMsgRateIn(quota.getMsgRateIn());
- shortTermData.setMsgRateOut(quota.getMsgRateOut());
- shortTermData.setMsgThroughputIn(quota.getBandwidthIn());
- shortTermData.setMsgThroughputOut(quota.getBandwidthOut());
-
- longTermData.setMsgRateIn(quota.getMsgRateIn());
- longTermData.setMsgRateOut(quota.getMsgRateOut());
- longTermData.setMsgThroughputIn(quota.getBandwidthIn());
- longTermData.setMsgThroughputOut(quota.getBandwidthOut());
-
- // Assume ample history.
- shortTermData.setNumSamples(NUM_SHORT_SAMPLES);
- longTermData.setNumSamples(NUM_LONG_SAMPLES);
- }
- } catch (Exception e) {
- log.warn("Error when trying to find bundle {} on metadata store: {}", bundle, e);
- }
- if (bundleData == null) {
- bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES, defaultStats);
- }
- return bundleData;
+ @Override
+ public CompletableFuture<BundleData> getBundleDataOrDefaultAsync(final String bundle) {
+ return pulsarResources.getLoadBalanceResources().getBundleDataResources().getBundleData(bundle)
+ .thenCompose(optBundleData -> {
+ if (optBundleData.isPresent()) {
+ return CompletableFuture.completedFuture(optBundleData.get());
+ }
+ return pulsarResources.getLoadBalanceResources().getQuotaResources().getQuota(bundle)
+ .thenApply(optQuota -> {
+ if (optQuota.isEmpty()) {
+ return null;
+ }
+ ResourceQuota quota = optQuota.get();
+ BundleData bundleData = new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES);
+ // Initialize from existing resource quotas if new API ZNodes do not exist.
+ final TimeAverageMessageData shortTermData = bundleData.getShortTermData();
+ final TimeAverageMessageData longTermData = bundleData.getLongTermData();
+
+ shortTermData.setMsgRateIn(quota.getMsgRateIn());
+ shortTermData.setMsgRateOut(quota.getMsgRateOut());
+ shortTermData.setMsgThroughputIn(quota.getBandwidthIn());
+ shortTermData.setMsgThroughputOut(quota.getBandwidthOut());
+
+ longTermData.setMsgRateIn(quota.getMsgRateIn());
+ longTermData.setMsgRateOut(quota.getMsgRateOut());
+ longTermData.setMsgThroughputIn(quota.getBandwidthIn());
+ longTermData.setMsgThroughputOut(quota.getBandwidthOut());
+
+ // Assume ample history.
+ shortTermData.setNumSamples(NUM_SHORT_SAMPLES);
+ longTermData.setNumSamples(NUM_LONG_SAMPLES);
+ return bundleData;
+ });
+ })
+ .exceptionally(e -> {
+ log.warn("Error when trying to find bundle {} on metadata store: {}", bundle, e);
+ return null;
+ })
+ .thenApply(bundleData -> bundleData != null ? bundleData
+ : new BundleData(NUM_SHORT_SAMPLES, NUM_LONG_SAMPLES, defaultStats));
}
// Use the Pulsar client to acquire the namespace bundle stats.
@@ -561,7 +570,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
} else {
// Otherwise, attempt to find the bundle data on metadata store.
// If it cannot be found, use the latest stats as the first sample.
- BundleData currentBundleData = getBundleDataOrDefault(bundle);
+ BundleData currentBundleData = getBundleDataOrDefaultAsync(bundle).join();
currentBundleData.update(stats);
bundleData.put(bundle, currentBundleData);
}
@@ -879,7 +888,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
private void preallocateBundle(String bundle, String broker) {
final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
+ key -> getBundleDataOrDefaultAsync(bundle).join());
loadData.getBrokerData().get(broker).getPreallocatedBundleData().put(bundle, data);
preallocatedBundleToBroker.put(bundle, broker);
@@ -893,7 +902,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
synchronized (brokerCandidateCache) {
final String bundle = serviceUnit.toString();
final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
+ key -> getBundleDataOrDefaultAsync(bundle).join());
brokerCandidateCache.clear();
LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
getAvailableBrokers(),
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
index 69f5208ce671..846f64c32c09 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java
@@ -56,6 +56,7 @@ import org.apache.pulsar.common.policies.data.LocalPolicies;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.stats.CacheMetricsCollector;
import org.apache.pulsar.common.util.Backoff;
+import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.metadata.api.Notification;
import org.apache.pulsar.policies.data.loadbalancer.BundleData;
import org.slf4j.Logger;
@@ -244,19 +245,25 @@ public class NamespaceBundleFactory {
public CompletableFuture<NamespaceBundle> getBundleWithHighestThroughputAsync(NamespaceName nsName) {
LoadManager loadManager = pulsar.getLoadManager().get();
if (loadManager instanceof ModularLoadManagerWrapper) {
- return getBundlesAsync(nsName).thenApply(bundles -> {
- double maxMsgThroughput = -1;
- NamespaceBundle bundleWithHighestThroughput = null;
- for (NamespaceBundle bundle : bundles.getBundles()) {
- BundleData bundleData = ((ModularLoadManagerWrapper) loadManager).getLoadManager()
- .getBundleDataOrDefault(bundle.toString());
- if (bundleData.getTopics() > 0
- && bundleData.getLongTermData().totalMsgThroughput() > maxMsgThroughput) {
- maxMsgThroughput = bundleData.getLongTermData().totalMsgThroughput();
- bundleWithHighestThroughput = bundle;
+ return getBundlesAsync(nsName).thenCompose(bundles -> {
+ List<NamespaceBundle> bundleList = bundles.getBundles();
+ List<CompletableFuture<BundleData>> bundleDataFutures = bundleList.stream()
+ .map(bundle -> ((ModularLoadManagerWrapper) loadManager).getLoadManager()
+ .getBundleDataOrDefaultAsync(bundle.toString()))
+ .toList();
+ return FutureUtil.waitForAll(bundleDataFutures).thenApply(__ -> {
+ double maxMsgThroughput = -1;
+ NamespaceBundle bundleWithHighestThroughput = null;
+ for (int i = 0; i < bundleList.size(); i++) {
+ BundleData bundleData = bundleDataFutures.get(i).join();
+ if (bundleData.getTopics() > 0
+ && bundleData.getLongTermData().totalMsgThroughput() > maxMsgThroughput) {
+ maxMsgThroughput = bundleData.getLongTermData().totalMsgThroughput();
+ bundleWithHighestThroughput = bundleList.get(i);
+ }
}
- }
- return bundleWithHighestThroughput;
+ return bundleWithHighestThroughput;
+ });
});
}
return getBundleWithHighestTopicsAsync(nsName);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
index f6061204b62a..60486c96a184 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImplTest.java
@@ -1026,7 +1026,7 @@ public class ModularLoadManagerImplTest {
// get the bundleData of the first bundle range.
// The default value of the bundleData be the same as resourceQuota because the resourceQuota is present.
- BundleData defaultBundleData = lm.getBundleDataOrDefault(namespaceBundle.toString());
+ BundleData defaultBundleData = lm.getBundleDataOrDefaultAsync(namespaceBundle.toString()).join();
TimeAverageMessageData shortTermData = defaultBundleData.getShortTermData();
TimeAverageMessageData longTermData = defaultBundleData.getLongTermData();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
index 6fcebff2b81a..7193810bec47 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceServiceTest.java
@@ -688,7 +688,7 @@ public class NamespaceServiceTest extends BrokerTestBase {
LoadManager loadManager = pulsar.getLoadManager().get();
Awaitility.await().untilAsserted(() -> {
BundleData targetBundleData = ((ModularLoadManagerWrapper) loadManager).getLoadManager()
- .getBundleDataOrDefault(namespace + "/" + bundle);
+ .getBundleDataOrDefaultAsync(namespace + "/" + bundle).join();
assertEquals(targetBundleData.getTopics(), 10);
});