You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/03/06 01:34:43 UTC
[pulsar] branch branch-2.11 updated: [fix][broker] Fix loadbalance score caculation problem (#19420)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new fdd88067812 [fix][broker] Fix loadbalance score caculation problem (#19420)
fdd88067812 is described below
commit fdd88067812ba61641905d52dcc3b45d2894587f
Author: gaozhangmin <zh...@apache.org>
AuthorDate: Wed Feb 15 17:53:37 2023 +0800
[fix][broker] Fix loadbalance score caculation problem (#19420)
---
.../loadbalance/impl/ModularLoadManagerImpl.java | 29 ++++++++++------------
1 file changed, 13 insertions(+), 16 deletions(-)
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 6579d35b335..1cd22e8233e 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
@@ -42,6 +42,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.SystemUtils;
@@ -62,6 +63,7 @@ import org.apache.pulsar.broker.resources.ClusterResources;
import org.apache.pulsar.broker.stats.prometheus.metrics.Summary;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.util.ExecutorProvider;
+import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceBundleFactory;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.ServiceUnitId;
@@ -576,24 +578,19 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
// Remove all loaded bundles from the preallocated maps.
final Map<String, BundleData> preallocatedBundleData = brokerData.getPreallocatedBundleData();
+ Set<String> ownedNsBundles = pulsar.getNamespaceService().getOwnedServiceUnits()
+ .stream().map(NamespaceBundle::toString).collect(Collectors.toSet());
synchronized (preallocatedBundleData) {
- for (String preallocatedBundleName : brokerData.getPreallocatedBundleData().keySet()) {
- if (brokerData.getLocalData().getBundles().contains(preallocatedBundleName)) {
- final Iterator<Map.Entry<String, BundleData>> preallocatedIterator =
- preallocatedBundleData.entrySet()
- .iterator();
- while (preallocatedIterator.hasNext()) {
- final String bundle = preallocatedIterator.next().getKey();
-
- if (bundleData.containsKey(bundle)) {
- preallocatedIterator.remove();
- preallocatedBundleToBroker.remove(bundle);
- }
- }
+ preallocatedBundleToBroker.keySet().removeAll(preallocatedBundleData.keySet());
+ final Iterator<Map.Entry<String, BundleData>> preallocatedIterator =
+ preallocatedBundleData.entrySet().iterator();
+ while (preallocatedIterator.hasNext()) {
+ final String bundle = preallocatedIterator.next().getKey();
+ if (!ownedNsBundles.contains(bundle)
+ || (brokerData.getLocalData().getBundles().contains(bundle)
+ && bundleData.containsKey(bundle))) {
+ preallocatedIterator.remove();
}
-
- // This is needed too in case a broker which was assigned a bundle dies and comes back up.
- preallocatedBundleToBroker.remove(preallocatedBundleName);
}
}