You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "void-ptr974 (via GitHub)" <gi...@apache.org> on 2026/07/26 07:51:39 UTC
[PR] [fix][broker] Use stable bundle names in AvgShedder [pulsar]
void-ptr974 opened a new pull request, #26246:
URL: https://github.com/apache/pulsar/pull/26246
### Motivation
AvgShedder retains a destination chosen during load shedding. Using mutable BundleData as the cache key can conflate equal bundle data and lose the destination when load data refreshes the object. It can also leave stale parent entries after a bundle is deleted or split.
### Modifications
- Cache planned destinations by canonical bundle name.
- Pass the bundle name through a backward-compatible placement-strategy overload.
- Invalidate cached destinations when a bundle is removed or successfully split.
- Preserve legacy selection behavior for existing strategy implementations and AvgShedder subclasses.
- Add regressions for value-equal BundleData, refreshed instances, split/delete cleanup, and legacy dispatch.
### Verification
- ./gradlew :pulsar-broker:checkstyleMain :pulsar-broker:checkstyleTest :pulsar-broker:compileJava :pulsar-broker:test --tests org.apache.pulsar.broker.loadbalance.impl.AvgShedderTest --tests org.apache.pulsar.broker.loadbalance.ModularLoadManagerStrategyTest
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3739523741
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -273,20 +288,39 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
@Override
public Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf) {
- final var brokerToUnload = bundleBrokerMap.getOrDefault(bundleToAssign, null);
- if (brokerToUnload == null || !candidates.contains(bundleBrokerMap.get(bundleToAssign))) {
- // cluster initializing or broker is shutdown
- if (!bundleBrokerMap.containsKey(bundleToAssign)) {
+ return candidates.isEmpty() ? Optional.empty() : Optional.of(getExpectedBroker(candidates, bundleToAssign));
+ }
+
+ @Override
+ public Optional<String> selectBrokerForBundle(Set<String> candidates, String bundle, BundleData bundleToAssign,
+ LoadData loadData, ServiceConfiguration conf) {
+ return bundle == null
+ ? selectBroker(candidates, bundleToAssign, loadData, conf)
+ : selectBrokerWithBundleName(candidates, bundle, bundleToAssign);
+ }
+
+ private Optional<String> selectBrokerWithBundleName(Set<String> candidates, String bundle,
+ BundleData bundleToAssign) {
+ if (candidates.isEmpty()) {
+ return Optional.empty();
+ }
+ final var pendingBroker = pendingBundleToBroker.get(bundle);
+ if (pendingBroker == null || !candidates.contains(pendingBroker)) {
+ // cluster initializing or the pending broker is shutdown or filtered out
+ if (pendingBroker == null) {
log.debug("cluster is initializing");
} else {
- log.debug().attr("broker", bundleBrokerMap.get(bundleToAssign)).attr("candidates", candidates)
+ log.debug().attr("broker", pendingBroker).attr("candidates", candidates)
.log("expected broker is shutdown");
}
String broker = getExpectedBroker(candidates, bundleToAssign);
- bundleBrokerMap.put(bundleToAssign, broker);
+ if (pendingBroker != null) {
+ // Keep a replacement only for the remainder of this unload attempt, including the retry path.
+ pendingBundleToBroker.put(bundle, broker);
Review Comment:
I handled this in the same update by removing the fallback write-back. A temporary fallback now leaves the pending plan unchanged, so it cannot reinsert the entry after completion. The test covers fallback, restoration of the original destination, and cleanup.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -981,7 +986,7 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
getAvailableBrokers(),
brokerTopicLoadingPredicate);
Optional<String> brokerTmp =
- placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
Thanks for the careful review. The retry now removes the first selected broker from the widened candidate set before selecting again. A regression test verifies that the retry cannot return the same overloaded broker.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategy.java:
##########
@@ -46,6 +46,26 @@ public interface ModularLoadManagerStrategy {
Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf);
+ /**
+ * Find a suitable broker to assign the given bundle when its stable name is available.
+ *
+ * @param candidates
+ * The candidates for which the bundle may be assigned.
+ * @param bundle
+ * The stable bundle name.
+ * @param bundleToAssign
+ * The data for the bundle to assign.
+ * @param loadData
+ * The load data from the leader broker.
+ * @param conf
+ * The service configuration.
+ * @return The name of the selected broker as it appears on ZooKeeper.
Review Comment:
Thanks, I’ve updated the Javadoc to use “broker ID” consistently.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
Review Comment:
This edge case is theoretically possible if a later broker pair fails after an earlier pair has populated the plan. No unload is issued in that case, the task is rescheduled, and the next AvgShedder pass clears the pending state. To keep this change focused, I suggest handling transactional publication separately if we can reproduce an assignment issue.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
+ bundlesToUnload.asMap().forEach((broker, bundles) -> {
+ AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
+ bundles.forEach(bundle -> {
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ if (sheddingExcludedNamespaces.contains(namespaceName)) {
+ log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("namespace", namespaceName)
+ .log("Skipping load shedding for namespace");
+ return;
+ }
+ if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
- if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
- return;
- }
- NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
- Optional<String> destBroker = this.selectBroker(bundleToUnload);
- if (!destBroker.isPresent()) {
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("broker", broker)
- .log("No broker available to unload bundle from broker");
- return;
- }
- if (destBroker.get().equals(broker)) {
- log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("broker", destBroker.get()).attr("bundle", bundle)
- .log("The destination broker is the same as the current owner broker for bundle");
- return;
- }
+ if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("broker", broker)
+ .log("No broker available to unload bundle from broker");
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("broker", destBroker.get()).attr("bundle", bundle)
+ .log("The destination broker is the same as the current owner broker for bundle");
+ return;
+ }
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
- .log("Unloading bundle from source broker to dest broker");
- try {
- pulsar.getAdminClient().namespaces()
- .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
- loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
- unloadBundleCount++;
- unloadBundleForBroker.set(true);
- } catch (PulsarServerException | PulsarAdminException e) {
- log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
- .log("Error when trying to perform load shedding on for broker");
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
+ .log("Unloading bundle from source broker to dest broker");
+ try {
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
+ loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
+ unloadBundleCount++;
+ unloadBundleForBroker.set(true);
+ } catch (PulsarServerException | PulsarAdminException e) {
+ log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
+ .log("Error when trying to perform load shedding on for broker");
+ }
+ });
+ if (unloadBundleForBroker.get()) {
+ unloadBrokerCount++;
}
});
- if (unloadBundleForBroker.get()) {
- unloadBrokerCount++;
- }
- });
+ } finally {
+ loadSheddingStrategy.onUnloadAttemptCompleted(plannedBundles);
+ }
Review Comment:
On this lifecycle point, the AvgShedder entry is only used to select the unload destination. Before the source is unloaded, the destination is stored separately in and later consumed by lookup. Therefore, this cleanup does not remove the target affinity, and the existing affinity tests cover this flow.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
+ bundlesToUnload.asMap().forEach((broker, bundles) -> {
+ AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
+ bundles.forEach(bundle -> {
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ if (sheddingExcludedNamespaces.contains(namespaceName)) {
+ log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("namespace", namespaceName)
+ .log("Skipping load shedding for namespace");
+ return;
+ }
+ if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
- if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
- return;
- }
- NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
- Optional<String> destBroker = this.selectBroker(bundleToUnload);
- if (!destBroker.isPresent()) {
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("broker", broker)
- .log("No broker available to unload bundle from broker");
- return;
- }
- if (destBroker.get().equals(broker)) {
- log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("broker", destBroker.get()).attr("bundle", bundle)
- .log("The destination broker is the same as the current owner broker for bundle");
- return;
- }
+ if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("broker", broker)
+ .log("No broker available to unload bundle from broker");
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("broker", destBroker.get()).attr("bundle", bundle)
+ .log("The destination broker is the same as the current owner broker for bundle");
+ return;
+ }
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
- .log("Unloading bundle from source broker to dest broker");
- try {
- pulsar.getAdminClient().namespaces()
- .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
- loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
- unloadBundleCount++;
- unloadBundleForBroker.set(true);
- } catch (PulsarServerException | PulsarAdminException e) {
- log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
- .log("Error when trying to perform load shedding on for broker");
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
+ .log("Unloading bundle from source broker to dest broker");
+ try {
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
+ loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
+ unloadBundleCount++;
+ unloadBundleForBroker.set(true);
+ } catch (PulsarServerException | PulsarAdminException e) {
+ log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
+ .log("Error when trying to perform load shedding on for broker");
+ }
+ });
+ if (unloadBundleForBroker.get()) {
+ unloadBrokerCount++;
}
});
- if (unloadBundleForBroker.get()) {
- unloadBrokerCount++;
- }
- });
+ } finally {
+ loadSheddingStrategy.onUnloadAttemptCompleted(plannedBundles);
+ }
Review Comment:
On this lifecycle point, the AvgShedder entry is only used to select the unload destination. Before the source is unloaded, the destination is stored separately in bundleBrokerAffinityMap and later consumed by lookup. Therefore, this cleanup does not remove the target affinity, and the existing affinity tests cover this flow.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3789113625
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -980,8 +985,9 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
getAvailableBrokers(),
brokerTopicLoadingPredicate);
+ brokerCandidateCache.remove(broker.get());
Review Comment:
Removing the first broker resolves AvgShedder’s pending-destination retry issue, but it also alters the overload-retry semantics for all ModularLoadManagerStrategy instances.
Previously, the behavior expanded the candidate set and allowed the placement strategy to determine whether a better broker existed. With this change, we now force a different broker even when all alternatives are worse.
For instance, with an overload threshold of 85%, if broker A is at 90% and B at 98%, removing A forces the selection of B. The LeastLongTermMessageRate strategy also treats any broker above the threshold as +INF, so once A is removed, it cannot distinguish between them.
Another issue is that the expanded candidate set is rebuilt only via applyNamespacePolicies(), while earlier filters—such as anti-affinity, topic-count, and broker filters—are not reapplied. If A was the only broker that passed those filters, removing it forces selection from brokers that were intentionally excluded. Overload handling has historically involved anti-affinity considerations (see #9393).
Could we instead adjust the retry to bypass AvgShedder’s pending destination specifically, rather than modifying candidate semantics for all placement strategies? For example, AvgShedder could ignore a pending destination once that broker is confirmed to be overloaded.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,23 +286,43 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * @deprecated This overload does not receive a stable bundle name and therefore cannot reuse a pending
+ * load-shedding destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData,
+ * ServiceConfiguration)} when the bundle name is available.
+ */
+ @Deprecated
@Override
public Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf) {
- final var brokerToUnload = bundleBrokerMap.getOrDefault(bundleToAssign, null);
- if (brokerToUnload == null || !candidates.contains(bundleBrokerMap.get(bundleToAssign))) {
- // cluster initializing or broker is shutdown
- if (!bundleBrokerMap.containsKey(bundleToAssign)) {
+ return candidates.isEmpty() ? Optional.empty() : Optional.of(getExpectedBroker(candidates, bundleToAssign));
+ }
+
+ @Override
+ public Optional<String> selectBrokerForBundle(Set<String> candidates, String bundle, BundleData bundleToAssign,
+ LoadData loadData, ServiceConfiguration conf) {
+ return bundle == null
+ ? selectBroker(candidates, bundleToAssign, loadData, conf)
+ : selectBrokerWithBundleName(candidates, bundle, bundleToAssign);
+ }
+
+ private Optional<String> selectBrokerWithBundleName(Set<String> candidates, String bundle,
+ BundleData bundleToAssign) {
+ if (candidates.isEmpty()) {
+ return Optional.empty();
+ }
+ final var pendingBroker = pendingBundleToBroker.get(bundle);
+ if (pendingBroker == null || !candidates.contains(pendingBroker)) {
Review Comment:
One compatibility consideration: existing subclasses of AvgShedder that only override the historical `selectBroker(...)` method will no longer intercept normal placement requests.
ModularLoadManagerImpl now invokes `selectBrokerForBundle(...)`, which AvgShedder overrides. When there is no pending plan, this path directly calls the private `getExpectedBroker(...)` logic, bypassing the virtual `selectBroker(...)` method.
For instance, a custom `MyAvgShedder extends AvgShedder` with an overridden four-argument selector functioned correctly before this change but is silently bypassed now.
Could the case without a pending plan instead delegate to `selectBroker(...)`, reserving the name-aware path only for an actual pending destination? That would preserve existing subclass fallback behavior while still addressing the mutable BundleData key issue.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
Review Comment:
Upon re-examining the current threading context, I believe there is a concrete way to reproduce the issue.
`updateAll()` is not fully synchronized. After `cleanupDeadBrokersData()` releases the manager monitor, `updateAllBrokerData()` can mutate `loadData.getBrokerData()` while `doLoadShedding()` is still running.
The `AvgShedder` first snapshots broker names in `calculateScoresAndSort()`, then later in `selectBundleForUnloading()` it dereferences:
```
loadData.getBrokerData().get(underloadedBroker).getLocalData()
```
for each pair.
As a result, an earlier pair can already publish entries into `pendingBundleToBroker`, and then a broker from a later pair could disappear, causing planning to fail before `findBundlesForUnloading()` returns. Because the manager’s try/finally block starts after that call, those entries are never cleaned up.
Since `LoadSheddingTask` only schedules the next run, the aborted plan remains externally visible until the next shedding pass.
I don’t think full transactional publication is required to resolve this. At a minimum, could `AvgShedder` clear `pendingBundleToBroker` on an exceptional exit from `findBundlesForUnloading()`?
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3977561678
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,24 +286,39 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * @deprecated This overload does not receive a stable bundle name and therefore cannot reuse a pending
+ * load-shedding destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData,
+ * ServiceConfiguration)} when the bundle name is available.
+ */
+ @Deprecated
Review Comment:
Thanks, the annotation is removed and the supported fallback remains documented in [`AvgShedder.java:289-298`](https://github.com/apache/pulsar/blob/1191348c234274da48c35315ca2dd582f84fdc88/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java#L289-L298). This addresses the comment.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -959,10 +963,10 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
if (sheddingExcludedNamespaces.contains(namespaceNameFromBundleName)) {
log.debug().attr("bundle", bundle).log("Use round robin broker selector for bundle");
broker = sheddingExcludedNamespaceSelectionStrategy
- .selectBroker(brokerCandidateCache, data, loadData, conf);
+ .selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
} else {
// Choose a broker among the potentially smaller filtered list, when possible
- broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ broker = placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
Thanks, this addresses the retry concern. [`AvgShedder.java:328-347`](https://github.com/apache/pulsar/blob/1191348c234274da48c35315ca2dd582f84fdc88/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java#L328-L347) excludes the source and limits alternatives to healthy brokers, and the manager keeps the bundle-aware callback on retry. The tests cover the healthy-source case and retaining the plan when no healthy alternative remains.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #26246:
URL: https://github.com/apache/pulsar/pull/26246
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3886850709
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,24 +287,63 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * This selector does not receive a stable bundle name and therefore cannot reuse a pending load-shedding
+ * destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData, ServiceConfiguration)} when
+ * the bundle name is available.
+ */
@Override
public Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf) {
- final var brokerToUnload = bundleBrokerMap.getOrDefault(bundleToAssign, null);
- if (brokerToUnload == null || !candidates.contains(bundleBrokerMap.get(bundleToAssign))) {
- // cluster initializing or broker is shutdown
- if (!bundleBrokerMap.containsKey(bundleToAssign)) {
- log.debug("cluster is initializing");
- } else {
- log.debug().attr("broker", bundleBrokerMap.get(bundleToAssign)).attr("candidates", candidates)
- .log("expected broker is shutdown");
+ return candidates.isEmpty() ? Optional.empty() : Optional.of(getExpectedBroker(candidates, bundleToAssign));
+ }
+
+ @Override
+ public Optional<String> selectBrokerForBundle(Set<String> candidates, String bundle, BundleData bundleToAssign,
+ LoadData loadData, ServiceConfiguration conf) {
+ return bundle == null
+ ? selectBroker(candidates, bundleToAssign, loadData, conf)
+ : selectBrokerWithBundleName(candidates, bundle, bundleToAssign, loadData, conf);
+ }
+
+ private Optional<String> selectBrokerWithBundleName(Set<String> candidates, String bundle,
+ BundleData bundleToAssign, LoadData loadData,
+ ServiceConfiguration conf) {
+ final var pendingBroker = pendingBundleToBroker.get(bundle);
+ if (pendingBroker == null) {
+ return selectBroker(candidates, bundleToAssign, loadData, conf);
+ }
+ if (!candidates.contains(pendingBroker)) {
+ log.debug().attr("broker", pendingBroker).attr("bundle", bundle).attr("candidates", candidates)
+ .log("Planned load-shedding destination is unavailable");
+ return selectBroker(candidates, bundleToAssign, loadData, conf);
+ }
+ final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage();
+ final Optional<Double> pendingBrokerScore = getBrokerScore(pendingBroker, loadData, conf);
+ if (pendingBrokerScore.isPresent() && pendingBrokerScore.get() > overloadThreshold) {
+ Set<String> healthyCandidates = new HashSet<>();
+ for (String candidate : candidates) {
+ if (!candidate.equals(pendingBroker)
+ && getBrokerScore(candidate, loadData, conf)
+ .filter(score -> score <= overloadThreshold).isPresent()) {
Review Comment:
The shedding source can still be included in the healthyCandidates set.
AvgShedder selects a source broker based on the relative score difference between the highest and lowest brokers in a pair—it does not require the source to exceed the loadBalancerBrokerOverloadedThresholdPercentage.
A concrete example demonstrates this:
```
source broker1 = 80
planned broker2 = 30
other broker3 = 50
AvgShedder plans broker1 -> broker2
broker2 later rises to 90
hard overload threshold = 85
```
The `healthyCandidates` set becomes {broker1, broker3}. Since the fallback `AvgShedder` selector chooses randomly, it could select broker1. The `doLoadShedding()` function then sees `destination == source` and bypasses the load shedding entirely.
To avoid this scenario, the new test raises broker1’s load to 90 before testing the overloaded destination. However, a broker does not need to be fully overloaded to serve as an `AvgShedder` source.
Could the pending plan retain both the source and destination brokers—for example, as `bundle -> (sourceBroker, destinationBroker)`—and explicitly exclude the source when constructing alternatives? A regression test with source at 80%, planned destination at 90%, and another broker at 50% would cover this situation?
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3837677262
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
Review Comment:
Addressed in d1329a0f70c. The try/finally now includes planning, and AvgShedder clears the entire attempt plan on failure. The exceptional path is covered by testLoadSheddingPassesBundleNameAndCompletesAttempt.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -980,8 +985,9 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
LoadManagerShared.applyNamespacePolicies(serviceUnit, policies, brokerCandidateCache,
getAvailableBrokers(),
brokerTopicLoadingPredicate);
+ brokerCandidateCache.remove(broker.get());
Review Comment:
Agreed. The retry no longer removes the first broker; it uses the historical selector to bypass the AvgShedder pending destination while preserving candidate semantics. Covered by testOverloadRetryBypassesBundleAwareSelectionWithoutRemovingCandidates.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -981,7 +986,7 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
getAvailableBrokers(),
brokerTopicLoadingPredicate);
Optional<String> brokerTmp =
- placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
Superseding my previous reply: the first broker is no longer removed. The retry now bypasses only the pending destination.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,23 +286,43 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * @deprecated This overload does not receive a stable bundle name and therefore cannot reuse a pending
+ * load-shedding destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData,
+ * ServiceConfiguration)} when the bundle name is available.
+ */
+ @Deprecated
@Override
public Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf) {
- final var brokerToUnload = bundleBrokerMap.getOrDefault(bundleToAssign, null);
- if (brokerToUnload == null || !candidates.contains(bundleBrokerMap.get(bundleToAssign))) {
- // cluster initializing or broker is shutdown
- if (!bundleBrokerMap.containsKey(bundleToAssign)) {
+ return candidates.isEmpty() ? Optional.empty() : Optional.of(getExpectedBroker(candidates, bundleToAssign));
+ }
+
+ @Override
+ public Optional<String> selectBrokerForBundle(Set<String> candidates, String bundle, BundleData bundleToAssign,
+ LoadData loadData, ServiceConfiguration conf) {
+ return bundle == null
+ ? selectBroker(candidates, bundleToAssign, loadData, conf)
+ : selectBrokerWithBundleName(candidates, bundle, bundleToAssign);
+ }
+
+ private Optional<String> selectBrokerWithBundleName(Set<String> candidates, String bundle,
+ BundleData bundleToAssign) {
+ if (candidates.isEmpty()) {
+ return Optional.empty();
+ }
+ final var pendingBroker = pendingBundleToBroker.get(bundle);
+ if (pendingBroker == null || !candidates.contains(pendingBroker)) {
Review Comment:
Addressed in d1329a0f70c. Without a usable pending destination, the bundle-aware path delegates to the historical virtual selector, preserving subclass dispatch.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
+ bundlesToUnload.asMap().forEach((broker, bundles) -> {
+ AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
+ bundles.forEach(bundle -> {
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ if (sheddingExcludedNamespaces.contains(namespaceName)) {
+ log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("namespace", namespaceName)
+ .log("Skipping load shedding for namespace");
+ return;
+ }
+ if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
- if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
- return;
- }
- NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
- Optional<String> destBroker = this.selectBroker(bundleToUnload);
- if (!destBroker.isPresent()) {
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("broker", broker)
- .log("No broker available to unload bundle from broker");
- return;
- }
- if (destBroker.get().equals(broker)) {
- log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("broker", destBroker.get()).attr("bundle", bundle)
- .log("The destination broker is the same as the current owner broker for bundle");
- return;
- }
+ if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("broker", broker)
+ .log("No broker available to unload bundle from broker");
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("broker", destBroker.get()).attr("bundle", bundle)
+ .log("The destination broker is the same as the current owner broker for bundle");
+ return;
+ }
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
- .log("Unloading bundle from source broker to dest broker");
- try {
- pulsar.getAdminClient().namespaces()
- .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
- loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
- unloadBundleCount++;
- unloadBundleForBroker.set(true);
- } catch (PulsarServerException | PulsarAdminException e) {
- log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
- .log("Error when trying to perform load shedding on for broker");
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
+ .log("Unloading bundle from source broker to dest broker");
+ try {
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
+ loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
+ unloadBundleCount++;
+ unloadBundleForBroker.set(true);
+ } catch (PulsarServerException | PulsarAdminException e) {
+ log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
+ .log("Error when trying to perform load shedding on for broker");
+ }
+ });
+ if (unloadBundleForBroker.get()) {
+ unloadBrokerCount++;
}
});
- if (unloadBundleForBroker.get()) {
- unloadBrokerCount++;
- }
- });
+ } finally {
+ loadSheddingStrategy.onUnloadAttemptCompleted(plannedBundles);
+ }
Review Comment:
Yes, this is intentional. The pending plan ends after destination selection; directed unload stores the target in the existing one-shot affinity entry consumed by the next lookup. The callback does not indicate that the destination has acquired ownership; the Javadoc now states this explicitly.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3887014328
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,24 +287,63 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * This selector does not receive a stable bundle name and therefore cannot reuse a pending load-shedding
+ * destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData, ServiceConfiguration)} when
+ * the bundle name is available.
+ */
@Override
public Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf) {
- final var brokerToUnload = bundleBrokerMap.getOrDefault(bundleToAssign, null);
- if (brokerToUnload == null || !candidates.contains(bundleBrokerMap.get(bundleToAssign))) {
- // cluster initializing or broker is shutdown
- if (!bundleBrokerMap.containsKey(bundleToAssign)) {
- log.debug("cluster is initializing");
- } else {
- log.debug().attr("broker", bundleBrokerMap.get(bundleToAssign)).attr("candidates", candidates)
- .log("expected broker is shutdown");
+ return candidates.isEmpty() ? Optional.empty() : Optional.of(getExpectedBroker(candidates, bundleToAssign));
+ }
+
+ @Override
+ public Optional<String> selectBrokerForBundle(Set<String> candidates, String bundle, BundleData bundleToAssign,
+ LoadData loadData, ServiceConfiguration conf) {
+ return bundle == null
+ ? selectBroker(candidates, bundleToAssign, loadData, conf)
+ : selectBrokerWithBundleName(candidates, bundle, bundleToAssign, loadData, conf);
+ }
+
+ private Optional<String> selectBrokerWithBundleName(Set<String> candidates, String bundle,
+ BundleData bundleToAssign, LoadData loadData,
+ ServiceConfiguration conf) {
+ final var pendingBroker = pendingBundleToBroker.get(bundle);
+ if (pendingBroker == null) {
+ return selectBroker(candidates, bundleToAssign, loadData, conf);
+ }
+ if (!candidates.contains(pendingBroker)) {
+ log.debug().attr("broker", pendingBroker).attr("bundle", bundle).attr("candidates", candidates)
+ .log("Planned load-shedding destination is unavailable");
+ return selectBroker(candidates, bundleToAssign, loadData, conf);
+ }
+ final double overloadThreshold = conf.getLoadBalancerBrokerOverloadedThresholdPercentage();
+ final Optional<Double> pendingBrokerScore = getBrokerScore(pendingBroker, loadData, conf);
+ if (pendingBrokerScore.isPresent() && pendingBrokerScore.get() > overloadThreshold) {
+ Set<String> healthyCandidates = new HashSet<>();
+ for (String candidate : candidates) {
+ if (!candidate.equals(pendingBroker)
+ && getBrokerScore(candidate, loadData, conf)
+ .filter(score -> score <= overloadThreshold).isPresent()) {
Review Comment:
Thanks, you’re right—the shedding source may still be below the hard overload threshold, so filtering by load alone was not sufficient. Addressed in `620b75a5401`: the pending plan now keeps both the source and destination, and explicitly excludes the source when the planned destination is unavailable or overloaded. The regression test covers the `80 / 90 / 50` scenario and verifies that fallback cannot select the source.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3653187048
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategy.java:
##########
@@ -46,6 +46,26 @@ public interface ModularLoadManagerStrategy {
Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf);
+ /**
+ * Find a suitable broker to assign the given bundle when its stable name is available.
+ *
+ * @param candidates
+ * The candidates for which the bundle may be assigned.
+ * @param bundle
+ * The stable bundle name.
+ * @param bundleToAssign
+ * The data for the bundle to assign.
+ * @param loadData
+ * The load data from the leader broker.
+ * @param conf
+ * The service configuration.
+ * @return The name of the selected broker as it appears on ZooKeeper.
Review Comment:
the name of the broker is called `brokerId`, returned by org.apache.pulsar.broker.PulsarService#getBrokerId
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3885816942
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -959,10 +963,10 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
if (sheddingExcludedNamespaces.contains(namespaceNameFromBundleName)) {
log.debug().attr("bundle", bundle).log("Use round robin broker selector for bundle");
broker = sheddingExcludedNamespaceSelectionStrategy
- .selectBroker(brokerCandidateCache, data, loadData, conf);
+ .selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
} else {
// Choose a broker among the potentially smaller filtered list, when possible
- broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ broker = placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
Thanks, that makes sense. I’ve updated the retry path to keep using `selectBrokerForBundle(...)`, so the load manager no longer bypasses bundle-aware strategy behavior. AvgShedder now handles an overloaded planned destination within that path by selecting from healthy alternatives.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -959,10 +963,10 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
if (sheddingExcludedNamespaces.contains(namespaceNameFromBundleName)) {
log.debug().attr("bundle", bundle).log("Use round robin broker selector for bundle");
broker = sheddingExcludedNamespaceSelectionStrategy
- .selectBroker(brokerCandidateCache, data, loadData, conf);
+ .selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
} else {
// Choose a broker among the potentially smaller filtered list, when possible
- broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ broker = placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
Thanks, you’re right that restoring all candidates before calling the random selector could select the same overloaded broker again, or an overloaded source. AvgShedder now delegates only with healthy alternatives; if none is available, it keeps the original plan rather than selecting another overloaded broker.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,24 +286,39 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * @deprecated This overload does not receive a stable bundle name and therefore cannot reuse a pending
+ * load-shedding destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData,
+ * ServiceConfiguration)} when the bundle name is available.
+ */
+ @Deprecated
Review Comment:
Good point. Since the four-argument method remains the supported fallback for implementations that do not need bundle context, marking it as deprecated would be misleading. I’ve removed the annotation and the related suppressions.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3880732707
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -959,10 +963,10 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
if (sheddingExcludedNamespaces.contains(namespaceNameFromBundleName)) {
log.debug().attr("bundle", bundle).log("Use round robin broker selector for bundle");
broker = sheddingExcludedNamespaceSelectionStrategy
- .selectBroker(brokerCandidateCache, data, loadData, conf);
+ .selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
} else {
// Choose a broker among the potentially smaller filtered list, when possible
- broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ broker = placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
**[DESIGN] using the legacy 4-argument selector as the overload-retry hook makes the escape a re-roll rather than a better choice**
I agree with @Denovo1998's open point here, and I want to add the concrete consequence for `AvgShedder` itself, since the retry a few lines below now calls `placementStrategy.selectBroker(...)` specifically to bypass the pending destination.
`AvgShedder.selectBroker` ends in `getExpectedBroker`, which hashes `String.valueOf(new Random().nextInt())` (AvgShedder.java:332) — it looks at neither the bundle nor `loadData`. So on the retry path the manager does not pick a *better* broker than the overloaded first choice; it picks a uniformly random one from the restored candidate set. Two ways that bites:
- the re-roll can land on the same broker that was just rejected as overloaded, so the escape hatch silently does nothing;
- it can land on the source broker, and then `doLoadShedding` hits the `"The destination broker is the same as the current owner broker"` branch and skips the unload for that bundle altogether — so an overloaded planned destination can now cancel the shed instead of redirecting it.
Before this PR the retry was effectively a no-op for `AvgShedder`, because both calls went through the same `selectBroker` and, whenever the plan entry was still reachable, returned the same broker. Swapping the second call to the legacy selector changed that from "keep the plan" to "randomise", which I do not think is the intent.
This is the same thing @Denovo1998 is asking for from the API side: an explicit signal (a retry flag, an `ignorePendingDestination` hook, or a documented parameter on `selectBrokerForBundle`) would let `AvgShedder` drop its pending entry *and* still run its normal placement, and would stop the legacy selector from being load-bearing for strategies that legitimately implement only `selectBrokerForBundle`. Worth resolving before merge, since `testOverloadRetryBypassesBundleAwareSelectionWithoutRemovingCandidates` now pins the split contract in place.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -270,24 +286,39 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
return pairs;
}
+ /**
+ * @deprecated This overload does not receive a stable bundle name and therefore cannot reuse a pending
+ * load-shedding destination. Use {@link #selectBrokerForBundle(Set, String, BundleData, LoadData,
+ * ServiceConfiguration)} when the bundle name is available.
+ */
+ @Deprecated
Review Comment:
**[NIT] @Deprecated on this override contradicts the compatibility contract the PR is restoring**
Small one: the interface method `ModularLoadManagerStrategy#selectBroker` is not deprecated, and this override is still very much on the normal path — `selectBrokerWithBundleName` delegates to it whenever there is no usable pending destination (AvgShedder.java:313-320), and `ModularLoadManagerImpl` calls it directly on the overload retry. It is also the exact method the PR now asks subclass authors to keep overriding, which is what `testAvgShedderWithoutPendingDestinationDelegatesToSubclassSelector` pins.
So the annotation points people away from the method they must implement, and it already costs two `@SuppressWarnings("deprecation")` in the new tests. The Javadoc above it says the useful part clearly enough — I would keep the Javadoc and drop `@Deprecated`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/ModularLoadManagerStrategy.java:
##########
@@ -46,6 +46,26 @@ public interface ModularLoadManagerStrategy {
Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf);
+ /**
+ * Find a suitable broker to assign the given bundle when its stable name is available.
+ *
+ * @param candidates
+ * The candidates for which the bundle may be assigned.
+ * @param bundle
+ * The stable bundle name.
+ * @param bundleToAssign
+ * The data for the bundle to assign.
+ * @param loadData
+ * The load data from the leader broker.
+ * @param conf
+ * The service configuration.
+ * @return The name of the selected broker as it appears on ZooKeeper.
Review Comment:
Thanks — the broker-ID wording is consistent across both selectors now.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#issuecomment-5460801953
Thanks again for the thoughtful review. I’ve updated the implementation to address these concerns while keeping the existing strategy contract compatible. Please take another look when you have a chance.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "void-ptr974 (via GitHub)" <gi...@apache.org>.
void-ptr974 commented on PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#issuecomment-5091749075
Thanks for the detailed review — addressed in follow-up commits.
- Changed `AvgShedder` to store pending destinations as `bundle name -> broker`.
- Added `selectBrokerForBundle(...)` and wired `ModularLoadManagerImpl` to pass the bundle name to the placement strategy.
- Added `onUnloadAttemptCompleted(...)`; `ModularLoadManagerImpl` calls it in `finally` after processing a shedding plan.
- Replaced the shared map with `ConcurrentHashMap`. The existing 4-argument selector remains as an uncached compatibility fallback.
- Added tests for mutable/equal `BundleData`, unavailable planned destinations, cleanup, and the three-broker unload flow.
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3712987473
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -981,7 +986,7 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
getAvailableBrokers(),
brokerTopicLoadingPredicate);
Optional<String> brokerTmp =
- placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
The escape path for overload still fails to work for AvgShedder. The first call may return a pending destination; the second call widens the candidate set, but AvgShedder continues to return the same pending broker as long as it remains a candidate.
Even if load data changes after the shedding plan and the pending broker now exceeds the hard overload threshold, we still select it even when a healthier broker is available. Restricting the map to a single unload attempt does not alter this behavior.
Could retrying explicitly bypass or invalidate the pending destination? It would also be valuable to add a test in which the planned broker becomes overloaded between planning and assignment.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
+ bundlesToUnload.asMap().forEach((broker, bundles) -> {
+ AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
+ bundles.forEach(bundle -> {
+ final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
+ final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+ if (sheddingExcludedNamespaces.contains(namespaceName)) {
+ log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("namespace", namespaceName)
+ .log("Skipping load shedding for namespace");
+ return;
+ }
+ if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
- if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
- return;
- }
- NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
- Optional<String> destBroker = this.selectBroker(bundleToUnload);
- if (!destBroker.isPresent()) {
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("broker", broker)
- .log("No broker available to unload bundle from broker");
- return;
- }
- if (destBroker.get().equals(broker)) {
- log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("broker", destBroker.get()).attr("bundle", bundle)
- .log("The destination broker is the same as the current owner broker for bundle");
- return;
- }
+ if (!shouldAntiAffinityNamespaceUnload(namespaceName, bundleRange, broker)) {
+ return;
+ }
+ NamespaceBundle bundleToUnload = LoadManagerShared.getNamespaceBundle(pulsar, bundle);
+ Optional<String> destBroker = this.selectBroker(bundleToUnload);
+ if (!destBroker.isPresent()) {
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("broker", broker)
+ .log("No broker available to unload bundle from broker");
+ return;
+ }
+ if (destBroker.get().equals(broker)) {
+ log.warn().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("broker", destBroker.get()).attr("bundle", bundle)
+ .log("The destination broker is the same as the current owner broker for bundle");
+ return;
+ }
- log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
- .log("Unloading bundle from source broker to dest broker");
- try {
- pulsar.getAdminClient().namespaces()
- .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
- loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
- unloadBundleCount++;
- unloadBundleForBroker.set(true);
- } catch (PulsarServerException | PulsarAdminException e) {
- log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
- .log("Error when trying to perform load shedding on for broker");
+ log.info().attr("class", loadSheddingStrategy.getClass().getSimpleName())
+ .attr("bundle", bundle).attr("sourceBroker", broker).attr("destBroker", destBroker.get())
+ .log("Unloading bundle from source broker to dest broker");
+ try {
+ pulsar.getAdminClient().namespaces()
+ .unloadNamespaceBundle(namespaceName, bundleRange, destBroker.get());
+ loadData.getRecentlyUnloadedBundles().put(bundle, System.currentTimeMillis());
+ unloadBundleCount++;
+ unloadBundleForBroker.set(true);
+ } catch (PulsarServerException | PulsarAdminException e) {
+ log.warn().attr("bundle", bundle).attr("broker", broker).exception(e)
+ .log("Error when trying to perform load shedding on for broker");
+ }
+ });
+ if (unloadBundleForBroker.get()) {
+ unloadBrokerCount++;
}
});
- if (unloadBundleForBroker.get()) {
- unloadBrokerCount++;
- }
- });
+ } finally {
+ loadSheddingStrategy.onUnloadAttemptCompleted(plannedBundles);
+ }
Review Comment:
This cleanup depends on the completion of the unload at the source, not on the completion of the target assignment.
For the legacy load manager, `OwnedBundle#handleUnloadRequest` finishes after closing topics and removing ownership. The destination affinity is later consumed by `ModularLoadManagerWrapper#getLeastLoaded` when a lookup actually occurs. As a result, a delayed lookup or destination failure can happen after this callback has removed the pending plan.
Is this intentionally outside the pending-plan contract? If the plan is meant to cover assignment retries, then cleanup should be driven by consumption or expiry, not by the return of the unload RPC.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/AvgShedder.java:
##########
@@ -273,20 +288,39 @@ private List<Pair<String, String>> findBrokerPairs(List<String> brokers,
@Override
public Optional<String> selectBroker(Set<String> candidates, BundleData bundleToAssign, LoadData loadData,
ServiceConfiguration conf) {
- final var brokerToUnload = bundleBrokerMap.getOrDefault(bundleToAssign, null);
- if (brokerToUnload == null || !candidates.contains(bundleBrokerMap.get(bundleToAssign))) {
- // cluster initializing or broker is shutdown
- if (!bundleBrokerMap.containsKey(bundleToAssign)) {
+ return candidates.isEmpty() ? Optional.empty() : Optional.of(getExpectedBroker(candidates, bundleToAssign));
+ }
+
+ @Override
+ public Optional<String> selectBrokerForBundle(Set<String> candidates, String bundle, BundleData bundleToAssign,
+ LoadData loadData, ServiceConfiguration conf) {
+ return bundle == null
+ ? selectBroker(candidates, bundleToAssign, loadData, conf)
+ : selectBrokerWithBundleName(candidates, bundle, bundleToAssign);
+ }
+
+ private Optional<String> selectBrokerWithBundleName(Set<String> candidates, String bundle,
+ BundleData bundleToAssign) {
+ if (candidates.isEmpty()) {
+ return Optional.empty();
+ }
+ final var pendingBroker = pendingBundleToBroker.get(bundle);
+ if (pendingBroker == null || !candidates.contains(pendingBroker)) {
+ // cluster initializing or the pending broker is shutdown or filtered out
+ if (pendingBroker == null) {
log.debug("cluster is initializing");
} else {
- log.debug().attr("broker", bundleBrokerMap.get(bundleToAssign)).attr("candidates", candidates)
+ log.debug().attr("broker", pendingBroker).attr("candidates", candidates)
.log("expected broker is shutdown");
}
String broker = getExpectedBroker(candidates, bundleToAssign);
- bundleBrokerMap.put(bundleToAssign, broker);
+ if (pendingBroker != null) {
+ // Keep a replacement only for the remainder of this unload attempt, including the retry path.
+ pendingBundleToBroker.put(bundle, broker);
Review Comment:
A cleanup race condition exists here. A selector may read `pendingBroker`, while `onUnloadAttemptCompleted` could remove the key, and this `put` might reinsert a replacement after the attempt has already finished.
While `ConcurrentHashMap` prevents structural corruption, it does not make this lifecycle transition atomic. Could we use a conditional replacement, such as:
```java
pendingBundleToBroker.replace(bundle, pendingBroker, broker)
```
and treat a failed replacement as the attempt already being closed? An attempt-id or state object would be even safer. A latch-based concurrency test could cover this race.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -650,58 +650,63 @@ public synchronized void doLoadShedding() {
Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
final Multimap<String, String> bundlesToUnload = loadSheddingStrategy.findBundlesForUnloading(loadData, conf);
+ final Set<String> plannedBundles = new HashSet<>(bundlesToUnload.values());
- bundlesToUnload.asMap().forEach((broker, bundles) -> {
- AtomicBoolean unloadBundleForBroker = new AtomicBoolean(false);
- bundles.forEach(bundle -> {
- final String namespaceName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
- final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
- if (sheddingExcludedNamespaces.contains(namespaceName)) {
- log.debug().attr("class", loadSheddingStrategy.getClass().getSimpleName())
- .attr("namespace", namespaceName)
- .log("Skipping load shedding for namespace");
- return;
- }
- if (!shouldNamespacePoliciesUnload(namespaceName, bundleRange, broker)) {
- return;
- }
+ try {
Review Comment:
The `try/finally` begins after `findBundlesForUnloading`, but AvgShedder modifies its shared pending map during that call. If planning throws after inserting one or more entries, `onUnloadAttemptCompleted` is never called, leaving a partial plan visible to placement.
Could AvgShedder instead construct the destinations in a local plan and only publish it after planning finishes successfully, or at least clear the partially built state when planning fails?
--
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] Fix AvgShedder assignment cache keying with stable bundle names [pulsar]
Posted by "Denovo1998 (via GitHub)" <gi...@apache.org>.
Denovo1998 commented on code in PR #26246:
URL: https://github.com/apache/pulsar/pull/26246#discussion_r3838537899
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java:
##########
@@ -959,10 +963,10 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
if (sheddingExcludedNamespaces.contains(namespaceNameFromBundleName)) {
log.debug().attr("bundle", bundle).log("Use round robin broker selector for bundle");
broker = sheddingExcludedNamespaceSelectionStrategy
- .selectBroker(brokerCandidateCache, data, loadData, conf);
+ .selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
} else {
// Choose a broker among the potentially smaller filtered list, when possible
- broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
+ broker = placementStrategy.selectBrokerForBundle(brokerCandidateCache, bundle, data, loadData, conf);
Review Comment:
This fixes the AvgShedder-specific retry without mutating the candidate set, but using the legacy 4-argument selector here creates a new abstraction leak.
`selectBrokerForBundle(...)` is introduced by this PR as the bundle-aware placement hook, and the stable bundle name remains available during retry. A placement strategy may legitimately override it because bundle identity is integral to its algorithm—such as consistent hashing—while keeping `selectBroker(...)` only as a compatibility fallback.
With the current implementation, such a strategy is silently bypassed whenever the first selection is overloaded. If its legacy selector returns `Optional.empty()`, the manager retains the original overloaded broker even though the bundle-aware selector could choose a better candidate.
The new regression test now codifies this split contract—`selectBrokerForBundle` once, then `selectBroker` on retry—so this is no longer just an AvgShedder implementation detail.
Could `ModularLoadManagerImpl` instead call `selectBrokerForBundle(...)` consistently and make the “ignore pending shedding destination” behavior explicit to AvgShedder? Alternatively, an explicit retry context or default hook would avoid using the legacy selector as a hidden `ignorePendingDestination` signal.
--
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