added loadBalancerServiceUnitStateMaxConcurrentOverrides conf

Author: heesung-sn

conf/broker.conf

@@ -1586,6 +1586,13 @@ loadManagerServiceUnitStateTableViewClassName=org.apache.pulsar.broker.loadbalan
 # enable it. It accepts `None` to disable it."
 loadBalancerServiceUnitTableViewSyncer=None
 
+
+# Specify the maximum number of concurrent orphan bundle ownership overrides.
+# The leader broker triggers these overrides upon detecting orphaned bundles.
+# It identifies orphan bundle ownerships by periodically scanning ownership data
+# and monitoring for broker shutdowns or inactive states.
+loadBalancerServiceUnitStateMaxConcurrentOverrides = 10
+
 ### --- Replication --- ###
 
 # Enable replication metrics

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

@@ -3023,6 +3023,16 @@ public double getLoadBalancerBandwidthOutResourceWeight() {
     )
     private ServiceUnitTableViewSyncerType loadBalancerServiceUnitTableViewSyncer = ServiceUnitTableViewSyncerType.None;
 
+    @FieldContext(
+            dynamic = true,
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "Specify the maximum number of concurrent orphan bundle ownership overrides. "
+                    + "The leader broker triggers these overrides upon detecting orphaned bundles. "
+                    + "It identifies orphan bundle ownerships by periodically scanning ownership data "
+                    + "and monitoring for broker shutdowns or inactive states."
+    )
+    private int loadBalancerServiceUnitStateMaxConcurrentOverrides = 10;
+
     /**** --- Replication. --- ****/
     @FieldContext(
         category = CATEGORY_REPLICATION,

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

@@ -106,7 +106,6 @@ public class ServiceUnitStateChannelImpl implements ServiceUnitStateChannel {
 
     private static final int OWNERSHIP_CLEAN_UP_MAX_WAIT_TIME_IN_MILLIS = 5000;
     private static final int OWNERSHIP_CLEAN_UP_WAIT_RETRY_DELAY_IN_MILLIS = 100;
-    private static final int MAX_CONCURRENT_OWNERSHIP_OVERRIDE = 500;
     public static final long VERSION_ID_INIT = 1; // initial versionId
     public static final long MAX_CLEAN_UP_DELAY_TIME_IN_SECS = 3 * 60; // 3 mins
     private static final long MIN_CLEAN_UP_DELAY_TIME_IN_SECS = 0; // 0 secs to clean immediately
@@ -1503,7 +1502,7 @@ private void doHealthCheckBrokerAsyncWithRetries(String brokerId, int retry, Com
     }
 
     private void tryWaitForOverrides(List<CompletableFuture<Void>> overrideFutures, boolean force) {
-        if (overrideFutures.size() > MAX_CONCURRENT_OWNERSHIP_OVERRIDE || force) {
+        if (overrideFutures.size() >= config.getLoadBalancerServiceUnitStateMaxConcurrentOverrides() || force) {
             try {
                 FutureUtil.waitForAll(overrideFutures)
                         .get(config.getMetadataStoreOperationTimeoutSeconds(), SECONDS);