Repository navigation
[fix][broker] Avoid infinite bundle unloading #20822
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,6 +18,7 @@ | |
| */ | ||
| package org.apache.pulsar.broker.loadbalance.impl; | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting; | ||
| import com.google.common.collect.Multimap; | ||
| import com.google.common.collect.Sets; | ||
| import java.util.ArrayList; | ||
|
|
@@ -659,11 +660,24 @@ 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.info("[{}] No broker available to unload bundle {} from broker {}", | ||
| strategy.getClass().getSimpleName(), bundle, broker); | ||
| return; | ||
| } | ||
| if (destBroker.get().equals(broker)) { | ||
| log.warn("[{}] The destination broker {} is the same as the current owner broker for Bundle {}", | ||
| strategy.getClass().getSimpleName(), destBroker.get(), bundle); | ||
| return; | ||
| } | ||
|
|
||
| log.info("[{}] Unloading bundle: {} from broker {} to dest broker {}", | ||
| strategy.getClass().getSimpleName(), bundle, broker, destBroker.get()); | ||
|
Demogorgon314 marked this conversation as resolved.
|
||
| try { | ||
| pulsar.getAdminClient().namespaces() | ||
| .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get()); | ||
|
Comment on lines
+679
to
+680
Member
Author
There was a problem hiding this comment. The
Contributor
There was a problem hiding this comment. Just a reminder, The
Contributor
There was a problem hiding this 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
Member
Author
There was a problem hiding this comment. When we have many brokers, it still has a chance to select the same broker if we don't pass the
Contributor
There was a problem hiding this 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.
Member
Author
There was a problem hiding this comment. Ok, I see. After this PR is merged, I will push a PR for branch 2.10.
Contributor
There was a problem hiding this 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?
Member
Author
There was a problem hiding this comment. @BewareMyPower It will be selected to unload again. And hope next time it will be skipped.
Contributor
There was a problem hiding this comment.
To be clear, how are you going to push this fix to 2.10? According to @codelipenghui, the |
||
| loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis()); | ||
| } catch (PulsarServerException | PulsarAdminException e) { | ||
| log.warn("Error when trying to perform load shedding on {} for broker {}", bundle, broker, e); | ||
|
|
@@ -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)); | ||
| } | ||
|
|
||
| Optional<String> broker = selectBroker(serviceUnit); | ||
| if (!broker.isPresent()) { | ||
| // If no broker is selected, return empty. | ||
| return broker; | ||
| } | ||
| // Add new bundle to preallocated. | ||
| preallocateBundle(bundle, broker.get()); | ||
| return broker; | ||
| } | ||
| } finally { | ||
| selectBrokerForAssignment.observe(System.nanoTime() - startTime, TimeUnit.NANOSECONDS); | ||
| } | ||
| } | ||
|
|
||
| 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); | ||
| } | ||
|
Comment on lines
+877
to
+887
Contributor
There was a problem hiding this comment. The
Member
Author
There was a problem hiding this comment. Yes... not sure why used |
||
| } | ||
|
|
||
| @VisibleForTesting | ||
| Optional<String> selectBroker(final ServiceUnitId serviceUnit) { | ||
| synchronized (brokerCandidateCache) { | ||
| final String bundle = serviceUnit.toString(); | ||
| 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, bundle, | ||
| brokerCandidateCache, | ||
| brokerToNamespaceToBundleRange, brokerToFailureDomainMap); | ||
|
|
||
| // distribute bundles evenly to candidate-brokers if enable | ||
| if (conf.isLoadBalancerDistributeBundlesEvenlyEnabled()) { | ||
| LoadManagerShared.removeMostServicingBrokersForNamespace(bundle, | ||
| brokerCandidateCache, | ||
| brokerToNamespaceToBundleRange); | ||
| if (log.isDebugEnabled()) { | ||
| log.debug("enable distribute bundles evenly to candidate-brokers, broker candidate count={}", | ||
| brokerCandidateCache.size()); | ||
| } | ||
| } | ||
|
|
||
| log.info("{} brokers being considered for assignment of {}", brokerCandidateCache.size(), bundle); | ||
|
|
||
| // 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); | ||
| } | ||
|
|
||
| if (brokerCandidateCache.isEmpty()) { | ||
| // restore the list of brokers to the full set | ||
| LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache, | ||
| getAvailableBrokers(), | ||
| brokerTopicLoadingPredicate); | ||
| } | ||
|
|
||
| // Choose a broker among the potentially smaller filtered list, when possible | ||
| Optional<String> broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf); | ||
| if (log.isDebugEnabled()) { | ||
| log.debug("Selected broker {} from candidate brokers {}", broker, brokerCandidateCache); | ||
| } | ||
|
|
||
| if (!broker.isPresent()) { | ||
| // No brokers available | ||
| return broker; | ||
| } | ||
|
|
||
| final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage() / 100.0; | ||
| final double maxUsage = loadData.getBrokerData().get(broker.get()).getLocalData().getMaxResourceUsage(); | ||
| if (maxUsage > overloadThreshold) { | ||
| // All brokers that were in the filtered list were overloaded, so check if there is a better broker | ||
| LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache, | ||
| getAvailableBrokers(), | ||
| brokerTopicLoadingPredicate); | ||
| Optional<String> brokerTmp = | ||
| placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf); | ||
| if (brokerTmp.isPresent()) { | ||
| broker = brokerTmp; | ||
| } | ||
| } | ||
| return broker; | ||
| } | ||
| } | ||
|
|
||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.