[improve][broker] Make AvgShedder the default load shedding and placement strategy

### Motivation

The modular load manager still defaults to ThresholdShedder for shedding and
LeastLongTermMessageRate for placement. The two score brokers differently
(resource usage vs. message rate), so a bundle shed from a hot broker is
routinely placed on another hot broker and shed again. ThresholdShedder's
leader-side EMA also lags actual load by minutes, which over-unloads after
scale-outs and rolling restarts, and it never sheds towards an idle broker
unless lowerBoundarySheddingEnabled is set. AvgShedder (PIP-364) pairs the
highest and lowest loaded brokers, pre-plans the destination of every bundle
it unloads, and only acts after a gap has persisted for several consecutive
checks, which removes the shed/place mismatch and the oscillation.

### Modifications

- ServiceConfiguration, conf/broker.conf, conf/standalone.conf:
  loadBalancerLoadSheddingStrategy and loadBalancerLoadPlacementStrategy
  default to AvgShedder; maxUnloadPercentage defaults to 0.5 so AvgShedder
  equalizes a broker pair in one cycle; loadBalancerDistributeBundlesEvenlyEnabled
  defaults to false and is now documented in broker.conf, since the
  per-namespace bundle-count filter runs before placement and can discard
  the destination AvgShedder planned for a bundle. standalone.conf gains the
  strategy keys and the AvgShedder settings it did not list.
- ModularLoadManagerImpl: strategy pairing is extracted into
  createLoadBalanceStrategies(). A configuration that sets a classic shedder
  (ThresholdShedder, UniformLoadShedder, OverloadShedder) while leaving the
  placement strategy at the new AvgShedder default no longer fails to start:
  the configured shedder is kept and placement falls back to
  LeastLongTermMessageRate with a warning, preserving the previous behavior
  for existing configurations. AvgShedder shedding with a different placement
  strategy logs a warning that its planned destinations are not honored.
- ModularLoadManagerImplStrategyTest covers the defaults, the fallback, an
  explicit classic pairing, and AvgShedder shedding with other placement.

Assisted-by: Claude Code (Opus 5)

Author: lhotari

conf/broker.conf

@@ -1755,14 +1755,33 @@ supportedNamespaceBundleSplitAlgorithms=range_equally_divide,topic_count_equally
 # Default algorithm name for namespace bundle split
 defaultNamespaceBundleSplitAlgorithm=range_equally_divide
 
-# load shedding strategy, support OverloadShedder and ThresholdShedder, default is ThresholdShedder since 2.10.0
-loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder
+# Load shedding strategy for the modular load manager (ModularLoadManagerImpl).
+# Supported: AvgShedder, ThresholdShedder, UniformLoadShedder and OverloadShedder.
+# Default is AvgShedder since 5.0.0 (ThresholdShedder was the default from 2.10.0 to 4.x).
+# AvgShedder implements both the shedding and the placement strategy: it pairs the highest and the lowest
+# loaded brokers, moves load between them once their score gap has exceeded a threshold for several
+# consecutive checks, and places the unloaded bundles on the broker it chose for them. It must be paired
+# with loadBalancerLoadPlacementStrategy=AvgShedder (see below).
+# When the extensible load manager (ExtensibleLoadManagerImpl) is used, set this to
+# org.apache.pulsar.broker.loadbalance.extensions.scheduler.TransferShedder instead.
+loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
 
 # If enabled, when current usage < average usage - threshold, the broker with the highest load will be triggered to unload.
 lowerBoundarySheddingEnabled=false
 
-# load balance placement strategy, support LeastLongTermMessageRate and LeastResourceUsageWithWeight
-loadBalancerLoadPlacementStrategy=org.apache.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate
+# Load balance placement strategy for the modular load manager.
+# Supported: AvgShedder, LeastLongTermMessageRate and LeastResourceUsageWithWeight.
+# Default is AvgShedder since 5.0.0 (LeastLongTermMessageRate before), which only takes effect together with
+# loadBalancerLoadSheddingStrategy=AvgShedder. If a different shedding strategy is configured, the broker falls
+# back to LeastLongTermMessageRate placement and logs a warning; set this key explicitly in that case
+# (LeastResourceUsageWithWeight is the recommended pairing for ThresholdShedder).
+loadBalancerLoadPlacementStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+# Whether to narrow the candidate brokers for a new bundle assignment to those owning the fewest bundles of the
+# same namespace before the placement strategy runs. This distributes bundle counts evenly per namespace but
+# overrides load-aware placement and can discard the destination AvgShedder planned for an unloaded bundle.
+# Disabled by default since 5.0.0 (it was enabled before). System namespace bundles are always distributed evenly.
+loadBalancerDistributeBundlesEvenlyEnabled=false
 
 # The broker resource usage threshold.
 # When the broker resource usage is greater than the pulsar cluster average resource usage,
@@ -1814,7 +1833,7 @@ loadBalancerBundleUnloadMinThroughputThreshold=10
 # Time to wait for the unloading of a namespace bundle
 namespaceBundleUnloadingTimeoutMs=60000
 
-# configuration for AvgShedder, a new shedding and placement strategy
+# Configuration for AvgShedder, the default shedding and placement strategy since 5.0.0
 # The low threshold for the difference between the highest and lowest loaded brokers.
 loadBalancerAvgShedderLowThreshold = 15
 
@@ -1827,10 +1846,10 @@ loadBalancerAvgShedderHitCountLowThreshold = 8
 # The number of times the high threshold is triggered before the bundle is unloaded.
 loadBalancerAvgShedderHitCountHighThreshold = 2
 
-# In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio.
-# For AvgShedder, recommend to set to 0.5, so that it will distribute the load evenly
-#  between the highest and lowest brokers.
-maxUnloadPercentage = 0.2
+# In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio: the share of the load
+# difference between the highest and the lowest loaded broker that is moved in one shedding cycle.
+# Default is 0.5 since 5.0.0 (0.2 before), which lets AvgShedder equalize the two brokers in a single cycle.
+maxUnloadPercentage = 0.5
 
 
 ### --- Load balancer extension --- ###

conf/standalone.conf

@@ -1201,6 +1201,47 @@ loadBalancerNamespaceBundleMaxBandwidthMbytes=100
 # maximum number of bundles in a namespace
 loadBalancerNamespaceMaximumBundles=128
 
+# Load shedding strategy for the modular load manager (ModularLoadManagerImpl).
+# Supported: AvgShedder, ThresholdShedder, UniformLoadShedder and OverloadShedder.
+# Default is AvgShedder since 5.0.0 (ThresholdShedder was the default from 2.10.0 to 4.x).
+# AvgShedder implements both the shedding and the placement strategy and must be paired with
+# loadBalancerLoadPlacementStrategy=AvgShedder (see below).
+loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+# If enabled, when current usage < average usage - threshold, the broker with the highest load will be triggered to unload.
+# It only takes effect in the ThresholdShedder strategy.
+lowerBoundarySheddingEnabled=false
+
+# Load balance placement strategy for the modular load manager.
+# Supported: AvgShedder, LeastLongTermMessageRate and LeastResourceUsageWithWeight.
+# Default is AvgShedder since 5.0.0 (LeastLongTermMessageRate before), which only takes effect together with
+# loadBalancerLoadSheddingStrategy=AvgShedder. If a different shedding strategy is configured, the broker falls
+# back to LeastLongTermMessageRate placement and logs a warning; set this key explicitly in that case.
+loadBalancerLoadPlacementStrategy=org.apache.pulsar.broker.loadbalance.impl.AvgShedder
+
+# Whether to narrow the candidate brokers for a new bundle assignment to those owning the fewest bundles of the
+# same namespace before the placement strategy runs. Disabled by default since 5.0.0 (it was enabled before)
+# because it overrides load-aware placement and can discard the destination AvgShedder planned for a bundle.
+loadBalancerDistributeBundlesEvenlyEnabled=false
+
+# Configuration for AvgShedder, the default shedding and placement strategy since 5.0.0
+# The low threshold for the difference between the highest and lowest loaded brokers.
+loadBalancerAvgShedderLowThreshold = 15
+
+# The high threshold for the difference between the highest and lowest loaded brokers.
+loadBalancerAvgShedderHighThreshold = 40
+
+# The number of times the low threshold is triggered before the bundle is unloaded.
+loadBalancerAvgShedderHitCountLowThreshold = 8
+
+# The number of times the high threshold is triggered before the bundle is unloaded.
+loadBalancerAvgShedderHitCountHighThreshold = 2
+
+# In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio: the share of the load
+# difference between the highest and the lowest loaded broker that is moved in one shedding cycle.
+# Default is 0.5 since 5.0.0 (0.2 before), which lets AvgShedder equalize the two brokers in a single cycle.
+maxUnloadPercentage = 0.5
+
 # The broker resource usage threshold.
 # When the broker resource usage is greater than the pulsar cluster average resource usage,
 # the threshold shedder will be triggered to offload bundles from the broker.

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -3028,9 +3028,12 @@ in the Admin API message inspection endpoints (getMessageById, peekNthMessage,
             dynamic = true,
             doc = "load balance load shedding strategy "
                 + "(It requires broker restart if value is changed using dynamic config). "
-                + "Default is ThresholdShedder since 2.10.0"
+                + "Default is AvgShedder since 5.0.0 (ThresholdShedder was the default from 2.10.0 to 4.x). "
+                + "AvgShedder implements both the shedding and the placement strategy and must be paired with "
+                + "loadBalancerLoadPlacementStrategy=AvgShedder; when a different shedding strategy is configured, "
+                + "an AvgShedder placement strategy falls back to LeastLongTermMessageRate."
     )
-    private String loadBalancerLoadSheddingStrategy = "org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder";
+    private String loadBalancerLoadSheddingStrategy = "org.apache.pulsar.broker.loadbalance.impl.AvgShedder";
 
     @FieldContext(
             category = CATEGORY_LOAD_BALANCER,
@@ -3041,10 +3044,15 @@ in the Admin API message inspection endpoints (getMessageById, peekNthMessage,
 
     @FieldContext(
             category = CATEGORY_LOAD_BALANCER,
-            doc = "load balance placement strategy"
+            doc = "load balance placement strategy. "
+                    + "Default is AvgShedder since 5.0.0 (LeastLongTermMessageRate before), which binds placement to "
+                    + "the AvgShedder shedding strategy so that unloaded bundles land on the broker the shedder "
+                    + "chose for them. It only takes effect together with "
+                    + "loadBalancerLoadSheddingStrategy=AvgShedder; with any other shedding strategy the broker "
+                    + "falls back to LeastLongTermMessageRate placement and logs a warning."
     )
     private String loadBalancerLoadPlacementStrategy =
-            "org.apache.pulsar.broker.loadbalance.impl.LeastLongTermMessageRate";
+            "org.apache.pulsar.broker.loadbalance.impl.AvgShedder";
 
     @FieldContext(
         dynamic = true,
@@ -3087,9 +3095,14 @@ in the Admin API message inspection endpoints (getMessageById, peekNthMessage,
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "enable/disable distribute bundles evenly"
+            doc = "Enable/disable distributing bundles evenly across brokers when a bundle is assigned. "
+                    + "When enabled, the candidate brokers for a new assignment are first narrowed to those "
+                    + "owning the fewest bundles of that namespace, before the placement strategy runs. This "
+                    + "overrides load-aware placement and can discard the destination the AvgShedder shedding "
+                    + "strategy planned for an unloaded bundle, so it is disabled by default since 5.0.0 "
+                    + "(it was enabled before). Bundles of the system namespace are always distributed evenly."
     )
-    private boolean loadBalancerDistributeBundlesEvenlyEnabled = true;
+    private boolean loadBalancerDistributeBundlesEvenlyEnabled = false;
 
     @FieldContext(
         category = CATEGORY_LOAD_BALANCER,
@@ -3194,11 +3207,12 @@ in the Admin API message inspection endpoints (getMessageById, peekNthMessage,
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio."
-                    + "For AvgShedder, recommend to set to 0.5, so that it will distribute the load "
-                    + "evenly between the highest and lowest brokers."
+            doc = "In the UniformLoadShedder and AvgShedder strategy, the maximum unload ratio: the share of "
+                    + "the load difference between the highest and the lowest loaded broker that is moved in one "
+                    + "shedding cycle. Default is 0.5 since 5.0.0 (0.2 before), which lets AvgShedder equalize "
+                    + "the load of the two brokers in a single cycle."
     )
-    private double maxUnloadPercentage = 0.2;
+    private double maxUnloadPercentage = 0.5;
 
     @FieldContext(
         dynamic = true,

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

@@ -250,7 +250,9 @@ public void initialize(final PulsarService pulsar) {
         defaultStats.msgRateIn = DEFAULT_MESSAGE_RATE;
         defaultStats.msgRateOut = DEFAULT_MESSAGE_RATE;
 
-        placementStrategy = ModularLoadManagerStrategy.create(conf);
+        final LoadBalanceStrategies strategies = createLoadBalanceStrategies(conf);
+        placementStrategy = strategies.placementStrategy();
+        loadSheddingStrategy = strategies.loadSheddingStrategy();
         sheddingExcludedNamespaceSelectionStrategy = new RoundRobinBrokerSelector();
         policies = new SimpleResourceAllocationPolicies(pulsar);
         filterPipeline.add(new BrokerLoadManagerClassFilter());
@@ -263,22 +265,53 @@ public void initialize(final PulsarService pulsar) {
                     executors.execute(
                             () -> LoadManagerShared.refreshBrokerToFailureDomainMap(pulsar, brokerToFailureDomainMap));
                 });
+    }
+
+    /**
+     * The placement strategy and the load shedding strategy resolved from the configuration.
+     */
+    record LoadBalanceStrategies(ModularLoadManagerStrategy placementStrategy,
+                                 LoadSheddingStrategy loadSheddingStrategy) {
+    }
 
+    /**
+     * Creates the placement and the load shedding strategy from the configuration and pairs them.
+     * <p>
+     * A placement strategy that is also a load shedding strategy (AvgShedder, the default) plans the destination of
+     * every bundle it unloads, so it is only used when the same class is configured as the shedding strategy. If a
+     * different shedding strategy is configured explicitly, the configured shedder is kept and placement falls back
+     * to {@link LeastLongTermMessageRate}, the default placement strategy before AvgShedder, so configurations that
+     * only set {@code loadBalancerLoadSheddingStrategy} keep the behavior they had.
+     */
+    @VisibleForTesting
+    static LoadBalanceStrategies createLoadBalanceStrategies(ServiceConfiguration conf) {
+        ModularLoadManagerStrategy placementStrategy = ModularLoadManagerStrategy.create(conf);
         if (placementStrategy instanceof LoadSheddingStrategy) {
-            // if the placement strategy is also a load shedding strategy
-            // we need to check two strategies are the same
-            if (!conf.getLoadBalancerLoadSheddingStrategy().equals(
-                    conf.getLoadBalancerLoadPlacementStrategy())) {
-                throw new IllegalArgumentException("The load shedding strategy: "
-                        + conf.getLoadBalancerLoadSheddingStrategy()
-                        + " can't work with the placement strategy: "
-                        + conf.getLoadBalancerLoadPlacementStrategy());
+            if (conf.getLoadBalancerLoadSheddingStrategy().equals(conf.getLoadBalancerLoadPlacementStrategy())) {
+                // bind the load shedding strategy and the placement strategy
+                return new LoadBalanceStrategies(placementStrategy, (LoadSheddingStrategy) placementStrategy);
             }
-            // bind the load shedding strategy and the placement strategy
-            loadSheddingStrategy = (LoadSheddingStrategy) placementStrategy;
-        } else {
-            loadSheddingStrategy = createLoadSheddingStrategy();
+            log.warn()
+                    .attr("sheddingStrategy", conf.getLoadBalancerLoadSheddingStrategy())
+                    .attr("placementStrategy", conf.getLoadBalancerLoadPlacementStrategy())
+                    .attr("fallbackPlacementStrategy", LeastLongTermMessageRate.class.getName())
+                    .log("The configured load shedding strategy cannot be paired with the placement strategy,"
+                            + " which requires the same class as the shedding strategy. Using the fallback"
+                            + " placement strategy instead. Set loadBalancerLoadPlacementStrategy explicitly"
+                            + " to choose the placement strategy");
+            return new LoadBalanceStrategies(new LeastLongTermMessageRate(), createLoadSheddingStrategy(conf));
+        }
+        LoadSheddingStrategy loadSheddingStrategy = createLoadSheddingStrategy(conf);
+        if (loadSheddingStrategy instanceof ModularLoadManagerStrategy) {
+            log.warn()
+                    .attr("sheddingStrategy", conf.getLoadBalancerLoadSheddingStrategy())
+                    .attr("placementStrategy", conf.getLoadBalancerLoadPlacementStrategy())
+                    .log("The load shedding strategy also implements bundle placement but a different"
+                            + " placement strategy is configured; bundles it unloads are placed by the"
+                            + " configured placement strategy, not where the shedder planned them. Set"
+                            + " loadBalancerLoadPlacementStrategy to the same class to pair them");
         }
+        return new LoadBalanceStrategies(placementStrategy, loadSheddingStrategy);
     }
 
     public void handleDataNotification(Notification t) {
@@ -328,7 +361,7 @@ private boolean isMetadataSessionConnected() {
         return lastMetadataSessionEvent != null && lastMetadataSessionEvent.isConnected();
     }
 
-    private LoadSheddingStrategy createLoadSheddingStrategy() {
+    private static LoadSheddingStrategy createLoadSheddingStrategy(ServiceConfiguration conf) {
         return Reflections.createInstance(conf.getLoadBalancerLoadSheddingStrategy(), LoadSheddingStrategy.class,
                 Thread.currentThread().getContextClassLoader());
     }