[improve][broker] PIP-192 made split handler idempotent (#19988)

Master Issue: #16691

### Motivation

Raising a PR to implement: #16691

### Modifications
This PR 
- Makes split handler idempotent .
- Makes Leader's orphan monitor keep trying to send split msg until finished.
- Select bundle boundaries at the SplitScheduler to have the same split boundaries for each Split handler retry.
- Adds a split condition to check if the parent's Splitting state has moved.
- Made Admin Unload command forceful to unload any bundles in invalid states.

Author: heesung-sn

conf/broker.conf

@@ -1455,7 +1455,7 @@ loadBalancerNamespaceBundleSplitConditionHitCountThreshold=3
 # the service-unit state channel when there are a large number of bundles.
 # minimum value = 30 secs
 # (only used in load balancer extension logics)
-loadBalancerServiceUnitStateCleanUpDelayTimeInSeconds=604800
+loadBalancerServiceUnitStateTombstoneDelayTimeInSeconds=3600
 
 
 ### --- Replication --- ###

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

@@ -2609,7 +2609,7 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
                     + "minimum value = 30 secs"
                     + "(only used in load balancer extension logics)"
     )
-    private long loadBalancerServiceUnitStateCleanUpDelayTimeInSeconds = 604800;
+    private long loadBalancerServiceUnitStateTombstoneDelayTimeInSeconds = 3600;
 
     @FieldContext(
             category = CATEGORY_LOAD_BALANCER,

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

@@ -449,7 +449,7 @@ public CompletableFuture<Void> unloadNamespaceBundleAsync(ServiceUnitId bundle,
                         log.warn(msg);
                         throw new IllegalArgumentException(msg);
                     }
-                    Unload unload = new Unload(sourceBroker, bundle.toString(), destinationBroker);
+                    Unload unload = new Unload(sourceBroker, bundle.toString(), destinationBroker, true);
                     UnloadDecision unloadDecision =
                             new UnloadDecision(unload, UnloadDecision.Label.Success, UnloadDecision.Reason.Admin);
                     return unloadAsync(unloadDecision,

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

@@ -45,10 +45,25 @@ public ServiceUnitStateData(ServiceUnitState state, String dstBroker, String sou
                 System.currentTimeMillis(), versionId);
     }
 
+    public ServiceUnitStateData(ServiceUnitState state, String dstBroker, String sourceBroker,
+                                Map<String, Optional<String>> splitServiceUnitToDestBroker, boolean force,
+                                long versionId) {
+        this(state, dstBroker, sourceBroker, splitServiceUnitToDestBroker, force,
+                System.currentTimeMillis(), versionId);
+    }
+
     public ServiceUnitStateData(ServiceUnitState state, String dstBroker, String sourceBroker, long versionId) {
         this(state, dstBroker, sourceBroker, null, false, System.currentTimeMillis(), versionId);
     }
 
+    public ServiceUnitStateData(ServiceUnitState state, String dstBroker, String sourceBroker, boolean force,
+                                long versionId) {
+        this(state, dstBroker, sourceBroker, null, force,
+                System.currentTimeMillis(), versionId);
+    }
+
+
+
     public ServiceUnitStateData(ServiceUnitState state, String dstBroker, long versionId) {
         this(state, dstBroker, null, null, false, System.currentTimeMillis(), versionId);
     }

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

@@ -30,12 +30,8 @@ public record Split(
 
     public Split {
         Objects.requireNonNull(serviceUnit);
-        if (splitServiceUnitToDestBroker != null && splitServiceUnitToDestBroker.size() != 2) {
+        if (splitServiceUnitToDestBroker == null || splitServiceUnitToDestBroker.size() != 2) {
             throw new IllegalArgumentException("Split service unit should be split into 2 service units.");
         }
     }
-
-    public Split(String serviceUnit, String sourceBroker) {
-        this(serviceUnit, sourceBroker, null);
-    }
 }

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

@@ -24,12 +24,18 @@
 /**
  * Defines the information required to unload or transfer a service unit(e.g. bundle).
  */
-public record Unload(String sourceBroker, String serviceUnit, Optional<String> destBroker) {
+public record Unload(String sourceBroker, String serviceUnit, Optional<String> destBroker, boolean force) {
     public Unload {
         Objects.requireNonNull(sourceBroker);
         Objects.requireNonNull(serviceUnit);
     }
+
+
     public Unload(String sourceBroker, String serviceUnit) {
-        this(sourceBroker, serviceUnit, Optional.empty());
+        this(sourceBroker, serviceUnit, Optional.empty(), false);
+    }
+
+    public Unload(String sourceBroker, String serviceUnit, Optional<String> destBroker) {
+        this(sourceBroker, serviceUnit, destBroker, false);
     }
 }

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

@@ -26,18 +26,23 @@
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Unknown;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.loadbalance.extensions.LoadManagerContext;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
 import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateChannelImpl;
 import org.apache.pulsar.broker.loadbalance.extensions.models.Split;
 import org.apache.pulsar.broker.loadbalance.extensions.models.SplitCounter;
 import org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision;
 import org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared;
 import org.apache.pulsar.common.naming.NamespaceBundleFactory;
+import org.apache.pulsar.common.naming.NamespaceBundleSplitAlgorithm;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
 
@@ -53,12 +58,14 @@ public class DefaultNamespaceBundleSplitStrategyImpl implements NamespaceBundleS
     private final Set<SplitDecision> decisionCache;
     private final Map<String, Integer> namespaceBundleCount;
     private final Map<String, Integer> splitConditionHitCounts;
+    private final Map<String, String> splittingBundles;
     private final SplitCounter counter;
 
     public DefaultNamespaceBundleSplitStrategyImpl(SplitCounter counter) {
         decisionCache = new HashSet<>();
         namespaceBundleCount = new HashMap<>();
         splitConditionHitCounts = new HashMap<>();
+        splittingBundles = new HashMap<>();
         this.counter = counter;
 
     }
@@ -67,6 +74,7 @@ public DefaultNamespaceBundleSplitStrategyImpl(SplitCounter counter) {
     public Set<SplitDecision> findBundlesToSplit(LoadManagerContext context, PulsarService pulsar) {
         decisionCache.clear();
         namespaceBundleCount.clear();
+        splittingBundles.clear();
         final ServiceConfiguration conf = pulsar.getConfiguration();
         int maxBundleCount = conf.getLoadBalancerNamespaceMaximumBundles();
         long maxBundleTopics = conf.getLoadBalancerNamespaceBundleMaxTopics();
@@ -78,6 +86,15 @@ public Set<SplitDecision> findBundlesToSplit(LoadManagerContext context, PulsarS
         boolean debug = log.isDebugEnabled() || conf.isLoadBalancerDebugModeEnabled();
         var channel = ServiceUnitStateChannelImpl.get(pulsar);
 
+        for (var etr : channel.getOwnershipEntrySet()) {
+            var eData = etr.getValue();
+            if (eData.state() == ServiceUnitState.Splitting) {
+                String bundle = etr.getKey();
+                final String bundleRange = LoadManagerShared.getBundleRangeFromBundleName(bundle);
+                splittingBundles.put(bundle, bundleRange);
+            }
+        }
+
         Map<String, NamespaceBundleStats> bundleStatsMap = pulsar.getBrokerService().getBundleStats();
         NamespaceBundleFactory namespaceBundleFactory =
                 pulsar.getNamespaceService().getNamespaceBundleFactory();
@@ -177,6 +194,27 @@ public Set<SplitDecision> findBundlesToSplit(LoadManagerContext context, PulsarS
                 continue;
             }
 
+            var ranges = bundleRange.split("_");
+            var foundSplittingBundle = false;
+            for (var etr : splittingBundles.entrySet()) {
+                var splittingBundle = etr.getKey();
+                if (splittingBundle.startsWith(namespace)) {
+                    var splittingBundleRange = etr.getValue();
+                    if (splittingBundleRange.startsWith(ranges[0])
+                            || splittingBundleRange.endsWith(ranges[1])) {
+                        if (debug) {
+                            log.info(String.format(CANNOT_SPLIT_BUNDLE_MSG
+                                    + " (parent) bundle:%s is in Splitting state.", bundle, splittingBundle));
+                        }
+                        foundSplittingBundle = true;
+                        break;
+                    }
+                }
+            }
+            if (foundSplittingBundle) {
+                continue;
+            }
+
             if (debug) {
                 log.info(String.format(
                         "Splitting bundle: %s. "
@@ -193,7 +231,43 @@ public Set<SplitDecision> findBundlesToSplit(LoadManagerContext context, PulsarS
                 ));
             }
             var decision = new SplitDecision();
-            decision.setSplit(new Split(bundle, context.brokerRegistry().getBrokerId()));
+            var namespaceService = pulsar.getNamespaceService();
+            var namespaceBundle = namespaceService.getNamespaceBundleFactory()
+                    .getBundle(namespaceName, bundleRange);
+            NamespaceBundleSplitAlgorithm algorithm =
+                    namespaceService.getNamespaceBundleSplitAlgorithmByName(
+                            conf.getDefaultNamespaceBundleSplitAlgorithm());
+            List<Long> splitBoundary = null;
+            try {
+                splitBoundary = namespaceService
+                        .getSplitBoundary(namespaceBundle,  null, algorithm)
+                        .get(conf.getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
+            } catch (Throwable e) {
+                counter.update(Failure, Unknown);
+                log.warn(String.format(CANNOT_SPLIT_BUNDLE_MSG + " Failed to get split boundaries.", bundle, e));
+                continue;
+            }
+            if (splitBoundary == null) {
+                counter.update(Failure, Unknown);
+                log.warn(String.format(CANNOT_SPLIT_BUNDLE_MSG + " The split boundaries is null.", bundle));
+                continue;
+            }
+            if (splitBoundary.size() != 1) {
+                counter.update(Failure, Unknown);
+                log.warn(String.format(CANNOT_SPLIT_BUNDLE_MSG + " The size of split boundaries is not 1. "
+                        + "splitBoundary:%s", bundle, splitBoundary));
+                continue;
+            }
+
+            var parentRange = namespaceBundle.getKeyRange();
+            var leftChildBundle = namespaceBundleFactory.getBundle(namespaceBundle.getNamespaceObject(),
+                    NamespaceBundleFactory.getRange(parentRange.lowerEndpoint(), splitBoundary.get(0)));
+            var rightChildBundle = namespaceBundleFactory.getBundle(namespaceBundle.getNamespaceObject(),
+                    NamespaceBundleFactory.getRange(splitBoundary.get(0), parentRange.upperEndpoint()));
+            Map<String, Optional<String>> splitServiceUnitToDestBroker = Map.of(
+                    leftChildBundle.getBundleRange(), Optional.empty(),
+                    rightChildBundle.getBundleRange(), Optional.empty());
+            decision.setSplit(new Split(bundle, context.brokerRegistry().getBrokerId(), splitServiceUnitToDestBroker));
             decision.succeed(reason);
             decisionCache.add(decision);
             int bundleNum = namespaceBundleCount.getOrDefault(namespace, 0);

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

@@ -967,9 +967,7 @@ void splitAndOwnBundleOnceAndRetry(NamespaceBundle bundle,
      */
     public CompletableFuture<Pair<NamespaceBundles, List<NamespaceBundle>>> getSplitBoundary(
             NamespaceBundle bundle, NamespaceBundleSplitAlgorithm nsBundleSplitAlgorithm, List<Long> boundaries) {
-        BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config);
-        CompletableFuture<List<Long>> splitBoundary =
-                nsBundleSplitAlgorithm.getSplitBoundary(bundleSplitOption);
+        CompletableFuture<List<Long>> splitBoundary = getSplitBoundary(bundle, boundaries, nsBundleSplitAlgorithm);
         return splitBoundary.thenCompose(splitBoundaries -> {
                     if (splitBoundaries == null || splitBoundaries.size() == 0) {
                         LOG.info("[{}] No valid boundary found in {} to split bundle {}",
@@ -981,6 +979,12 @@ public CompletableFuture<Pair<NamespaceBundles, List<NamespaceBundle>>> getSplit
                 });
     }
 
+    public CompletableFuture<List<Long>> getSplitBoundary(
+            NamespaceBundle bundle, List<Long> boundaries, NamespaceBundleSplitAlgorithm nsBundleSplitAlgorithm) {
+        BundleSplitOption bundleSplitOption = getBundleSplitOption(bundle, boundaries, config);
+        return nsBundleSplitAlgorithm.getSplitBoundary(bundleSplitOption);
+    }
+
     private BundleSplitOption getBundleSplitOption(NamespaceBundle bundle,
                                                    List<Long> boundaries,
                                                    ServiceConfiguration config) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -272,8 +272,7 @@ public NamespaceBundle getBundle(String namespace, String bundleRange) {
         String[] boundaries = bundleRange.split("_");
         Long lowerEndpoint = Long.decode(boundaries[0]);
         Long upperEndpoint = Long.decode(boundaries[1]);
-        Range<Long> hashRange = Range.range(lowerEndpoint, BoundType.CLOSED, upperEndpoint,
-                (upperEndpoint.equals(NamespaceBundles.FULL_UPPER_BOUND)) ? BoundType.CLOSED : BoundType.OPEN);
+        Range<Long> hashRange = getRange(lowerEndpoint, upperEndpoint);
         return getBundle(NamespaceName.get(namespace), hashRange);
     }
 
@@ -414,4 +413,9 @@ public static String getNamespaceFromPoliciesPath(String path) {
         return Joiner.on("/").join(i);
     }
 
+    public static Range<Long> getRange(Long lowerEndpoint, Long upperEndpoint) {
+        return Range.range(lowerEndpoint, BoundType.CLOSED, upperEndpoint,
+                (upperEndpoint.equals(NamespaceBundles.FULL_UPPER_BOUND)) ? BoundType.CLOSED : BoundType.OPEN);
+    }
+
 }

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java

@@ -39,6 +39,9 @@ public interface NamespaceBundleSplitAlgorithm {
     NamespaceBundleSplitAlgorithm TOPIC_COUNT_EQUALLY_DIVIDE_ALGO = new TopicCountEquallyDivideBundleSplitAlgorithm();
     NamespaceBundleSplitAlgorithm SPECIFIED_POSITIONS_DIVIDE_ALGO =
             new SpecifiedPositionsBundleSplitAlgorithm();
+
+    NamespaceBundleSplitAlgorithm SPECIFIED_POSITIONS_DIVIDE_FORCE_ALGO =
+            new SpecifiedPositionsBundleSplitAlgorithm(true);
     NamespaceBundleSplitAlgorithm FLOW_OR_QPS_EQUALLY_DIVIDE_ALGO = new FlowOrQpsEquallyDivideBundleSplitAlgorithm();
 
     static NamespaceBundleSplitAlgorithm of(String algorithmName) {

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleSplitAlgorithm.java

@@ -29,6 +29,16 @@
  * This algorithm divides the bundle into several parts by the specified positions.
  */
 public class SpecifiedPositionsBundleSplitAlgorithm implements NamespaceBundleSplitAlgorithm{
+
+    private boolean force;
+
+    public SpecifiedPositionsBundleSplitAlgorithm() {
+        force = false;
+    }
+
+    public SpecifiedPositionsBundleSplitAlgorithm(boolean force) {
+        this.force = force;
+    }
     @Override
     public CompletableFuture<List<Long>> getSplitBoundary(BundleSplitOption bundleSplitOption) {
         NamespaceService service = bundleSplitOption.getService();
@@ -39,19 +49,28 @@ public CompletableFuture<List<Long>> getSplitBoundary(BundleSplitOption bundleSp
         }
         // sort all positions
         Collections.sort(positions);
-        return service.getOwnedTopicListForNamespaceBundle(bundle).thenCompose(topics -> {
-            if (topics == null || topics.size() <= 1) {
-                return CompletableFuture.completedFuture(null);
-            }
-            List<Long> splitBoundaries = positions
-                    .stream()
-                    .filter(position -> position > bundle.getLowerEndpoint() && position < bundle.getUpperEndpoint())
-                    .collect(Collectors.toList());
-
-            if (splitBoundaries.size() == 0) {
-                return CompletableFuture.completedFuture(null);
-            }
-            return CompletableFuture.completedFuture(splitBoundaries);
-        });
+        if (force) {
+            return getBoundaries(bundle, positions);
+        } else {
+            return service.getOwnedTopicListForNamespaceBundle(bundle)
+                    .thenCompose(topics -> {
+                        if (topics == null || topics.size() <= 1) {
+                            return CompletableFuture.completedFuture(null);
+                        }
+                        return getBoundaries(bundle, positions);
+                    });
+        }
+    }
+
+    private CompletableFuture<List<Long>> getBoundaries(NamespaceBundle bundle, List<Long> positions) {
+        List<Long> splitBoundaries = positions
+                .stream()
+                .filter(position -> position > bundle.getLowerEndpoint() && position < bundle.getUpperEndpoint())
+                .collect(Collectors.toList());
+
+        if (splitBoundaries.size() == 0) {
+            return CompletableFuture.completedFuture(null);
+        }
+        return CompletableFuture.completedFuture(splitBoundaries);
     }
 }

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/SpecifiedPositionsBundleSplitAlgorithm.java

@@ -133,6 +133,7 @@ public void setup() throws Exception {
         conf.setLoadManagerClassName(ExtensibleLoadManagerImpl.class.getName());
         conf.setLoadBalancerLoadSheddingStrategy(TransferShedder.class.getName());
         conf.setLoadBalancerSheddingEnabled(false);
+        conf.setLoadBalancerDebugModeEnabled(true);
         super.internalSetup(conf);
         pulsar1 = pulsar;
         ServiceConfiguration defaultConf = getDefaultConf();