Addressed comments

Author: Demogorgon314

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

@@ -2924,9 +2924,9 @@ public double getLoadBalancerBandwidthOutResourceWeight() {
     @FieldContext(
             dynamic = true,
             category = CATEGORY_LOAD_BALANCER,
-            doc = "The namespaces skip for load shedding"
+            doc = "The namespaces to be excluded from load shedding"
     )
-    private Set<String> loadBalancerSheddingExcludedNamespaces = new TreeSet<>();
+    private Set<String> loadBalancerSheddingExcludedNamespaces = new HashSet<>();
 
     @FieldContext(
             category = CATEGORY_LOAD_BALANCER,

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java

@@ -162,7 +162,7 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager, BrokerS
     @Getter
     private final BrokerSelectionStrategy brokerSelectionStrategy;
 
-    private final BrokerSelectionStrategy roundRobinBrokerSelectionStrategy;
+    private final BrokerSelectionStrategy sheddingExcludedNamespaceSelectionStrategy;
 
     @Getter
     private final List<BrokerFilter> brokerFilterPipeline;
@@ -257,7 +257,7 @@ public ExtensibleLoadManagerImpl() {
         this.brokerFilterPipeline.add(new BrokerMaxTopicCountFilter());
         this.brokerFilterPipeline.add(new BrokerVersionFilter());
         this.brokerSelectionStrategy = createBrokerSelectionStrategy();
-        this.roundRobinBrokerSelectionStrategy = new RoundRobinBrokerSelectionStrategy();
+        this.sheddingExcludedNamespaceSelectionStrategy = new RoundRobinBrokerSelectionStrategy();
     }
 
     public static boolean isLoadManagerExtensionEnabled(PulsarService pulsar) {
@@ -640,20 +640,33 @@ public CompletableFuture<Optional<String>> selectAsync(ServiceUnitId bundle,
                             return Optional.empty();
                         }
                         Set<String> candidateBrokers = availableBrokerCandidates.keySet();
-                        Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
-
-                        var namespace = NamespaceBundle.getBundleNamespace(bundle.toString());
-                        if (sheddingExcludedNamespaces.contains(namespace)) {
-                            if (debug(conf, log)) {
-                                log.info("Use round robin broker selector for {}", bundle);
-                            }
-                            return roundRobinBrokerSelectionStrategy.select(candidateBrokers, bundle, context);
-                        }
-                        return getBrokerSelectionStrategy().select(candidateBrokers, bundle, context);
+                        return getBrokerSelectionStrategy(bundle).select(candidateBrokers, bundle, context);
                     });
                 });
     }
 
+    /**
+     * For shedding excluded namespaces, use RoundRobinBrokerSelector to assign the ownership,
+     * it can make the assignment more average because these will not automatically rebalance to
+     * another broker unless manually unloaded it.
+     *
+     * @param bundle the bundle to assign
+     * @return the broker selection strategy
+     */
+    private BrokerSelectionStrategy getBrokerSelectionStrategy(ServiceUnitId bundle) {
+
+        Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
+
+        var namespace = NamespaceBundle.getBundleNamespace(bundle.toString());
+        if (sheddingExcludedNamespaces.contains(namespace)) {
+            if (debug(conf, log)) {
+                log.info("Use round robin broker selector for {}", bundle);
+            }
+            return sheddingExcludedNamespaceSelectionStrategy;
+        }
+        return brokerSelectionStrategy;
+    }
+
     @Override
     public CompletableFuture<Boolean> checkOwnershipAsync(Optional<ServiceUnitId> topic, ServiceUnitId bundleUnit) {
         return getOwnershipAsync(topic, bundleUnit)

pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/scheduler/TransferShedder.java

@@ -504,7 +504,8 @@ public Set<UnloadDecision> findBundlesForUnloading(LoadManagerContext context,
                     final String namespaceName = NamespaceBundle.getBundleNamespace(bundle);
                     if (sheddingExcludedNamespaces.contains(namespaceName)) {
                         if (debugMode) {
-                            log.info("Skipping load shedding for namespace {}", namespaceName);
+                            log.info(String.format(CANNOT_UNLOAD_BUNDLE_MSG
+                                    + " Bundle namespace has been found in sheddingExcludedNamespaces", bundle));
                         }
                         continue;
                     }

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

@@ -151,7 +151,7 @@ public class ModularLoadManagerImpl implements ModularLoadManager {
     // Strategy used to determine where new topics should be placed.
     private ModularLoadManagerStrategy placementStrategy;
 
-    private ModularLoadManagerStrategy roundRobinBrokerSelector;
+    private ModularLoadManagerStrategy sheddingExcludedNamespaceSelectionStrategy;
 
     // Policies used to determine which brokers are available for particular namespaces.
     private SimpleResourceAllocationPolicies policies;
@@ -253,7 +253,7 @@ public void initialize(final PulsarService pulsar) {
         defaultStats.msgRateOut = DEFAULT_MESSAGE_RATE;
 
         placementStrategy = ModularLoadManagerStrategy.create(conf);
-        roundRobinBrokerSelector = new RoundRobinBrokerSelector();
+        sheddingExcludedNamespaceSelectionStrategy = new RoundRobinBrokerSelector();
         policies = new SimpleResourceAllocationPolicies(pulsar);
         filterPipeline.add(new BrokerLoadManagerClassFilter());
         filterPipeline.add(new BrokerVersionFilter());
@@ -926,13 +926,17 @@ Optional<String> selectBroker(final ServiceUnitId serviceUnit) {
             }
 
             Optional<String> broker;
+            // For shedding excluded namespaces, use RoundRobinBrokerSelector to assign the ownership,
+            // it can make the assignment more average because these will not automatically rebalance to
+            // another broker unless manually unloaded it.
             Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
             String namespaceNameFromBundleName = LoadManagerShared.getNamespaceNameFromBundleName(bundle);
             if (sheddingExcludedNamespaces.contains(namespaceNameFromBundleName)) {
                 if (log.isDebugEnabled()) {
                     log.debug("Use round robin broker selector for {}", bundle);
                 }
-                broker = roundRobinBrokerSelector.selectBroker(brokerCandidateCache, data, loadData, conf);
+                broker = sheddingExcludedNamespaceSelectionStrategy
+                        .selectBroker(brokerCandidateCache, data, loadData, conf);
             } else {
                 // Choose a broker among the potentially smaller filtered list, when possible
                 broker = placementStrategy.selectBroker(brokerCandidateCache, data, loadData, conf);
@@ -1143,7 +1147,15 @@ public void writeBrokerDataOnZooKeeper(boolean force) {
      */
     private int selectTopKBundle() {
         bundleArr.clear();
-        bundleArr.addAll(loadData.getBundleData().entrySet());
+        Set<String> sheddingExcludedNamespaces = conf.getLoadBalancerSheddingExcludedNamespaces();
+        for (Map.Entry<String, BundleData> entry : loadData.getBundleData().entrySet()) {
+            String bundle = entry.getKey();
+            String namespace = NamespaceBundle.getBundleNamespace(bundle);
+            if (sheddingExcludedNamespaces.contains(namespace)) {
+                continue;
+            }
+            bundleArr.add(entry);
+        }
 
         int maxNumberOfBundlesInBundleLoadReport = pulsar.getConfiguration()
                 .getLoadBalancerMaxNumberOfBundlesInBundleLoadReport();