You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Demogorgon314 (via GitHub)" <gi...@apache.org> on 2023/07/17 10:03:36 UTC
[GitHub] [pulsar] Demogorgon314 opened a new pull request, #20822: [improve][broker] Avoid assigning bundle to the previous unloaded broker
Demogorgon314 opened a new pull request, #20822:
URL: https://github.com/apache/pulsar/pull/20822
### Motivation
When unload happens, the bundle ownership assignment logic doesn't know the previous unloaded broker. It might assign the bundle to the same broker. In this case, it might cause an infinite bundle unloading loop.
### Modifications
* Avoid assigning bundle to the previous unloaded broker
### Verifying this change
See the `ModularLoadManagerImplTest#testAvoidAssignBundleToPreviousUnloadedBroker` unit test.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] heesung-sn commented on pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1654525885
Though this is another great improvement in MoudlarLoadManager, this is a behavior change too.
For this,
- Do we need to update any doc?
- Don't we need to put this logic under any config, as this is a behavior change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267745818
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
But it will only sometimes go to the same broker, not always. In 2.10, we don't have this API to support a strict owner assignment at the loading stage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 closed pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 closed pull request #20822: [fix][broker] Avoid infinite bundle unloading
URL: https://github.com/apache/pulsar/pull/20822
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1268071510
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -837,99 +851,119 @@ public Optional<String> selectBrokerForAssignment(final ServiceUnitId serviceUni
// If the given bundle is already in preallocated, return the selected broker.
return Optional.of(preallocatedBundleToBroker.get(bundle));
}
- final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
- brokerCandidateCache.clear();
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- // filter brokers which owns topic higher than threshold
- LoadManagerShared.filterBrokersWithLargeTopicCount(brokerCandidateCache, loadData,
- conf.getLoadBalancerBrokerMaxTopics());
-
- // distribute namespaces to domain and brokers according to anti-affinity-group
- LoadManagerShared.filterAntiAffinityGroupOwnedBrokers(pulsar, serviceUnit.toString(),
- brokerCandidateCache,
- brokerToNamespaceToBundleRange, brokerToFailureDomainMap);
-
- // distribute bundles evenly to candidate-brokers if enable
- if (conf.isLoadBalancerDistributeBundlesEvenlyEnabled()) {
- LoadManagerShared.removeMostServicingBrokersForNamespace(serviceUnit.toString(),
- brokerCandidateCache,
- brokerToNamespaceToBundleRange);
- if (log.isDebugEnabled()) {
- log.debug("enable distribute bundles evenly to candidate-brokers, broker candidate count={}",
- brokerCandidateCache.size());
- }
+ Optional<String> broker = selectBroker(serviceUnit);
+ if (!broker.isPresent()) {
+ // If no broker is selected, return empty.
+ return broker;
}
- log.info("{} brokers being considered for assignment of {}", brokerCandidateCache.size(), bundle);
+ // Add new bundle to preallocated.
+ preallocateBundle(bundle, broker.get());
+ return broker;
+ }
+ } finally {
+ selectBrokerForAssignment.observe(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
+ }
+ }
- // Use the filter pipeline to finalize broker candidates.
- try {
- for (BrokerFilter filter : filterPipeline) {
- filter.filter(brokerCandidateCache, data, loadData, conf);
- }
- } catch (BrokerFilterException x) {
- // restore the list of brokers to the full set
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- }
+ private void preallocateBundle(String bundle, String broker) {
+ final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
+ key -> getBundleDataOrDefault(bundle));
+ loadData.getBrokerData().get(broker).getPreallocatedBundleData().put(bundle, data);
+ preallocatedBundleToBroker.put(bundle, broker);
+
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ final ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<String>> namespaceToBundleRange =
+ brokerToNamespaceToBundleRange
+ .computeIfAbsent(broker,
+ k -> ConcurrentOpenHashMap.<String,
+ ConcurrentOpenHashSet<String>>newBuilder()
+ .build());
+ synchronized (namespaceToBundleRange) {
+ namespaceToBundleRange.computeIfAbsent(namespaceName,
+ k -> ConcurrentOpenHashSet.<String>newBuilder().build())
+ .add(bundleRange);
+ }
+ }
- if (brokerCandidateCache.isEmpty()) {
- // restore the list of brokers to the full set
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- }
+ @VisibleForTesting
+ protected Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
Review Comment:
```suggestion
Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
```
`protected` is not needed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267737261
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
Review Comment:
Yes, I added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1654817723
@heesung-sn
> Do we need to update any doc?
Yes, It's good to add a doc to notify the users.
> Do we need to put this logic under any config, as this is a behavior change?
I think we can leave config as is, since unloading to the same broker is not make sense.
> However, could infinite unloading still happen if unloading bounces to multiple brokers?
Yes, this can happen, not sure if we have a good way to prevent it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #20822:
URL: https://github.com/apache/pulsar/pull/20822
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1268093394
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
@BewareMyPower It will be selected to unload again. And hope next time it will be skipped.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid infinite bundle unloading [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1773729590
cherry-picked branch-2.11 by https://github.com/apache/pulsar/pull/20878
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267677128
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
Review Comment:
```suggestion
if (destBroker.isEmpty()) {
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
Review Comment:
```suggestion
log.warn("The destination broker {} is the same as the current owner broker for Bundle {}",
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
Review Comment:
Is it better to print the strategy class name for the above two logs?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
Or maybe we don't need to ensure the bundle will actually go to the dest broker? We just want to avoid the infinite loop as possible
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
Review Comment:
```suggestion
log.info("No broker available to unload bundle {} from broker {}", bundle, broker);
```
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
+ this.preallocateBundle(bundle, destBroker.get());
Review Comment:
Maybe we can skip it for the unloading stage? It should be fine if the load manager gets the wrong owner at the unloading stage for a short period of time.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
Just a reminder, The `destBroker` is only available after 2.11.0. We are not able to cherry-pick to branch-2.10
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid infinite bundle unloading [pulsar]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1729306572
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -837,99 +851,119 @@ public Optional<String> selectBrokerForAssignment(final ServiceUnitId serviceUni
// If the given bundle is already in preallocated, return the selected broker.
return Optional.of(preallocatedBundleToBroker.get(bundle));
}
- final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
- brokerCandidateCache.clear();
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- // filter brokers which owns topic higher than threshold
- LoadManagerShared.filterBrokersWithLargeTopicCount(brokerCandidateCache, loadData,
- conf.getLoadBalancerBrokerMaxTopics());
-
- // distribute namespaces to domain and brokers according to anti-affinity-group
- LoadManagerShared.filterAntiAffinityGroupOwnedBrokers(pulsar, serviceUnit.toString(),
- brokerCandidateCache,
- brokerToNamespaceToBundleRange, brokerToFailureDomainMap);
-
- // distribute bundles evenly to candidate-brokers if enable
- if (conf.isLoadBalancerDistributeBundlesEvenlyEnabled()) {
- LoadManagerShared.removeMostServicingBrokersForNamespace(serviceUnit.toString(),
- brokerCandidateCache,
- brokerToNamespaceToBundleRange);
- if (log.isDebugEnabled()) {
- log.debug("enable distribute bundles evenly to candidate-brokers, broker candidate count={}",
- brokerCandidateCache.size());
- }
+ Optional<String> broker = selectBroker(serviceUnit);
+ if (!broker.isPresent()) {
+ // If no broker is selected, return empty.
+ return broker;
}
- log.info("{} brokers being considered for assignment of {}", brokerCandidateCache.size(), bundle);
+ // Add new bundle to preallocated.
+ preallocateBundle(bundle, broker.get());
+ return broker;
+ }
+ } finally {
+ selectBrokerForAssignment.observe(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
+ }
+ }
- // Use the filter pipeline to finalize broker candidates.
- try {
- for (BrokerFilter filter : filterPipeline) {
- filter.filter(brokerCandidateCache, data, loadData, conf);
- }
- } catch (BrokerFilterException x) {
- // restore the list of brokers to the full set
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- }
+ private void preallocateBundle(String bundle, String broker) {
+ final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
+ key -> getBundleDataOrDefault(bundle));
+ loadData.getBrokerData().get(broker).getPreallocatedBundleData().put(bundle, data);
+ preallocatedBundleToBroker.put(bundle, broker);
+
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ final ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<String>> namespaceToBundleRange =
+ brokerToNamespaceToBundleRange
+ .computeIfAbsent(broker,
+ k -> ConcurrentOpenHashMap.<String,
+ ConcurrentOpenHashSet<String>>newBuilder()
+ .build());
+ synchronized (namespaceToBundleRange) {
+ namespaceToBundleRange.computeIfAbsent(namespaceName,
+ k -> ConcurrentOpenHashSet.<String>newBuilder().build())
+ .add(bundleRange);
+ }
Review Comment:
The `synchronized` block here seems meaningless?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267668452
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
The `destinationBroker` parameter is added in this PR: https://github.com/apache/pulsar/pull/18663
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267742054
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
+ this.preallocateBundle(bundle, destBroker.get());
Review Comment:
Make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid infinite bundle unloading [pulsar]
Posted by "hanmz (via GitHub)" <gi...@apache.org>.
hanmz commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1956466155
Why not call `selectBrokerForAssignment` directly, but add a new `selectBroker` method?The only difference I see is that it is not added to preallocated. What is the impact if added to preallocated?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1268064961
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
For 2.10 we don't have a good way for that. What will happen if the same broker is still selected after unloading? Will it retry?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267722291
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
When we have many brokers, it still has a chance to select the same broker if we don't pass the `destBroker ` parameter. I wonder if this is acceptable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on pull request #20822: [improve][broker] Avoid assigning bundle to the previous unloaded broker
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1641264496
> BTW, I noticed no one put data to `LoadData.brokerData` and `LoadData.bundleData` except for the test. Is it still useful? or maybe I missed something.
@codelipenghui Yes, it is still in use. The `ModularLoadManagerImpl#updateAllBrokerData` method puts data to `LoadData.brokerData`, and the `ModularLoadManagerImpl#updateBundleData` method puts data to `LoadData.bundleData`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] heesung-sn commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "heesung-sn (via GitHub)" <gi...@apache.org>.
heesung-sn commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1272432261
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
> Ok, I see. After this PR is merged, I will push a PR for branch 2.10.
To be clear, how are you going to push this fix to 2.10? According to @codelipenghui, the `destBroker` option in the unload admin is only available since 2.11.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1649964591
@Demogorgon314 Could you please help cherry-pick the PR into release branches?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid infinite bundle unloading [pulsar]
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1729669398
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -837,99 +851,119 @@ public Optional<String> selectBrokerForAssignment(final ServiceUnitId serviceUni
// If the given bundle is already in preallocated, return the selected broker.
return Optional.of(preallocatedBundleToBroker.get(bundle));
}
- final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
- key -> getBundleDataOrDefault(bundle));
- brokerCandidateCache.clear();
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- // filter brokers which owns topic higher than threshold
- LoadManagerShared.filterBrokersWithLargeTopicCount(brokerCandidateCache, loadData,
- conf.getLoadBalancerBrokerMaxTopics());
-
- // distribute namespaces to domain and brokers according to anti-affinity-group
- LoadManagerShared.filterAntiAffinityGroupOwnedBrokers(pulsar, serviceUnit.toString(),
- brokerCandidateCache,
- brokerToNamespaceToBundleRange, brokerToFailureDomainMap);
-
- // distribute bundles evenly to candidate-brokers if enable
- if (conf.isLoadBalancerDistributeBundlesEvenlyEnabled()) {
- LoadManagerShared.removeMostServicingBrokersForNamespace(serviceUnit.toString(),
- brokerCandidateCache,
- brokerToNamespaceToBundleRange);
- if (log.isDebugEnabled()) {
- log.debug("enable distribute bundles evenly to candidate-brokers, broker candidate count={}",
- brokerCandidateCache.size());
- }
+ Optional<String> broker = selectBroker(serviceUnit);
+ if (!broker.isPresent()) {
+ // If no broker is selected, return empty.
+ return broker;
}
- log.info("{} brokers being considered for assignment of {}", brokerCandidateCache.size(), bundle);
+ // Add new bundle to preallocated.
+ preallocateBundle(bundle, broker.get());
+ return broker;
+ }
+ } finally {
+ selectBrokerForAssignment.observe(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
+ }
+ }
- // Use the filter pipeline to finalize broker candidates.
- try {
- for (BrokerFilter filter : filterPipeline) {
- filter.filter(brokerCandidateCache, data, loadData, conf);
- }
- } catch (BrokerFilterException x) {
- // restore the list of brokers to the full set
- LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
- getAvailableBrokers(),
- brokerTopicLoadingPredicate);
- }
+ private void preallocateBundle(String bundle, String broker) {
+ final BundleData data = loadData.getBundleData().computeIfAbsent(bundle,
+ key -> getBundleDataOrDefault(bundle));
+ loadData.getBrokerData().get(broker).getPreallocatedBundleData().put(bundle, data);
+ preallocatedBundleToBroker.put(bundle, broker);
+
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ final ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<String>> namespaceToBundleRange =
+ brokerToNamespaceToBundleRange
+ .computeIfAbsent(broker,
+ k -> ConcurrentOpenHashMap.<String,
+ ConcurrentOpenHashSet<String>>newBuilder()
+ .build());
+ synchronized (namespaceToBundleRange) {
+ namespaceToBundleRange.computeIfAbsent(namespaceName,
+ k -> ConcurrentOpenHashSet.<String>newBuilder().build())
+ .add(bundleRange);
+ }
Review Comment:
Yes... not sure why used `synchronized` block here before.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Avoid infinite bundle unloading [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1773724866
hi @Demogorgon314 Can you help push a PR cherry-pick this change to branch-2.11?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267747787
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.warn("No broker available to unload bundle {} from broker {}", bundle, broker);
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn("Bundle {} destination broker {} is the same as the source broker {}",
+ bundle, destBroker.get(), broker);
+ return;
+ }
- log.info("[{}] Unloading bundle: {} from broker {}",
- strategy.getClass().getSimpleName(), bundle, broker);
+ log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}",
+ strategy.getClass().getSimpleName(), bundle, broker, destBroker.get());
try {
- pulsar.getAdminClient().namespaces().unloadNamespaceBundle(namespaceName, bundleRange);
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
Review Comment:
Ok, I see. After this PR is merged, I will push a PR for branch 2.10.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] Demogorgon314 commented on a diff in pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "Demogorgon314 (via GitHub)" <gi...@apache.org>.
Demogorgon314 commented on code in PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#discussion_r1267725085
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
Review Comment:
I see the branch-2.10 is required Java [JDK 11](https://adoptium.net/?variant=openjdk11) or [JDK 8](https://adoptium.net/?variant=openjdk8), so I’m using the `isPresent ` method here.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -659,12 +660,25 @@ public synchronized void doLoadShedding() {
if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
return;
}
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
Review Comment:
I see the branch-2.10 is required Java [JDK 11](https://adoptium.net/?variant=openjdk11) or [JDK 8](https://adoptium.net/?variant=openjdk8), so I’m using the `isPresent ` method here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codecov-commenter commented on pull request #20822: [fix][broker] Avoid infinite bundle unloading
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20822:
URL: https://github.com/apache/pulsar/pull/20822#issuecomment-1645385386
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/20822?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#20822](https://app.codecov.io/gh/apache/pulsar/pull/20822?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (531e964) into [master](https://app.codecov.io/gh/apache/pulsar/commit/57fbee446aebb5c9f4db3bed559cb7bef3749731?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (57fbee4) will **decrease** coverage by `0.34%`.
> The diff coverage is `77.33%`.
[](https://app.codecov.io/gh/apache/pulsar/pull/20822?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #20822 +/- ##
============================================
- Coverage 72.97% 72.64% -0.34%
- Complexity 32157 32227 +70
============================================
Files 1868 1856 -12
Lines 139164 139139 -25
Branches 15314 15329 +15
============================================
- Hits 101555 101073 -482
- Misses 29562 30002 +440
- Partials 8047 8064 +17
```
| Flag | Coverage Δ | |
|---|---|---|
| inttests | `24.18% <48.00%> (?)` | |
| systests | `?` | |
| unittests | `72.40% <76.00%> (+0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://app.codecov.io/gh/apache/pulsar/pull/20822?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...roker/loadbalance/impl/ModularLoadManagerImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/20822?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9sb2FkYmFsYW5jZS9pbXBsL01vZHVsYXJMb2FkTWFuYWdlckltcGwuamF2YQ==) | `82.75% <77.33%> (+0.27%)` | :arrow_up: |
... and [169 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/20822/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org