[improve][broker] PIP-192 Added SplitScheduler and DefaultNamespaceBundleSplitStrategyImpl (#19622)

Master Issue: #16691

### Motivation

We will start raising PRs to implement PIP-192, #16691

### Modifications

This PR implemented 
- SplitScheduler
- DefaultNamespaceBundleSplitStrategyImpl
- SplitManager
-  and their unit test.

Author: heesung-sn

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

@@ -2557,6 +2557,30 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
                     + "(only used in load balancer extension logics)"
     )
     private double loadBalancerBundleLoadReportPercentage = 10;
+    @FieldContext(
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "Service units'(bundles) split interval. Broker periodically checks whether "
+                    + "some service units(e.g. bundles) should split if they become hot-spots. "
+                    + "(only used in load balancer extension logics)"
+    )
+    private int loadBalancerSplitIntervalMinutes = 1;
+    @FieldContext(
+            category = CATEGORY_LOAD_BALANCER,
+            dynamic = true,
+            doc = "Max number of bundles to split to per cycle. "
+                    + "(only used in load balancer extension logics)"
+    )
+    private int loadBalancerMaxNumberOfBundlesToSplitPerCycle = 10;
+    @FieldContext(
+            category = CATEGORY_LOAD_BALANCER,
+            dynamic = true,
+            doc = "Threshold to the consecutive count of fulfilled split conditions. "
+                    + "If the split scheduler consecutively finds bundles that meet split conditions "
+                    + "many times bigger than this threshold, the scheduler will trigger splits on the bundles "
+                    + "(if the number of bundles is less than loadBalancerNamespaceMaximumBundles). "
+                    + "(only used in load balancer extension logics)"
+    )
+    private int loadBalancerNamespaceBundleSplitConditionThreshold = 5;
 
     @FieldContext(
             category = CATEGORY_LOAD_BALANCER,

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

@@ -45,16 +45,17 @@
 import org.apache.pulsar.broker.loadbalance.extensions.filter.BrokerIsolationPoliciesFilter;
 import org.apache.pulsar.broker.loadbalance.extensions.filter.BrokerMaxTopicCountFilter;
 import org.apache.pulsar.broker.loadbalance.extensions.filter.BrokerVersionFilter;
+import org.apache.pulsar.broker.loadbalance.extensions.manager.SplitManager;
 import org.apache.pulsar.broker.loadbalance.extensions.manager.UnloadManager;
 import org.apache.pulsar.broker.loadbalance.extensions.models.AssignCounter;
 import org.apache.pulsar.broker.loadbalance.extensions.models.SplitCounter;
-import org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision;
 import org.apache.pulsar.broker.loadbalance.extensions.models.Unload;
 import org.apache.pulsar.broker.loadbalance.extensions.models.UnloadCounter;
 import org.apache.pulsar.broker.loadbalance.extensions.models.UnloadDecision;
 import org.apache.pulsar.broker.loadbalance.extensions.reporter.BrokerLoadDataReporter;
 import org.apache.pulsar.broker.loadbalance.extensions.reporter.TopBundleLoadDataReporter;
 import org.apache.pulsar.broker.loadbalance.extensions.scheduler.LoadManagerScheduler;
+import org.apache.pulsar.broker.loadbalance.extensions.scheduler.SplitScheduler;
 import org.apache.pulsar.broker.loadbalance.extensions.scheduler.UnloadScheduler;
 import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStore;
 import org.apache.pulsar.broker.loadbalance.extensions.store.LoadDataStoreException;
@@ -103,7 +104,6 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
 
     @Getter
     private final List<BrokerFilter> brokerFilterPipeline;
-
     /**
      * The load data reporter.
      */
@@ -113,9 +113,12 @@ public class ExtensibleLoadManagerImpl implements ExtensibleLoadManager {
 
     private ScheduledFuture brokerLoadDataReportTask;
     private ScheduledFuture topBundlesLoadDataReportTask;
+    private SplitScheduler splitScheduler;
 
     private UnloadManager unloadManager;
 
+    private SplitManager splitManager;
+
     private boolean started = false;
 
     private final AssignCounter assignCounter = new AssignCounter();
@@ -166,7 +169,9 @@ public void start() throws PulsarServerException {
         this.serviceUnitStateChannel = new ServiceUnitStateChannelImpl(pulsar);
         this.brokerRegistry.start();
         this.unloadManager = new UnloadManager();
+        this.splitManager = new SplitManager(splitCounter);
         this.serviceUnitStateChannel.listen(unloadManager);
+        this.serviceUnitStateChannel.listen(splitManager);
         this.serviceUnitStateChannel.start();
 
         try {
@@ -184,7 +189,6 @@ public void start() throws PulsarServerException {
                 .brokerLoadDataStore(brokerLoadDataStore)
                 .topBundleLoadDataStore(topBundlesLoadDataStore).build();
 
-
         this.brokerLoadDataReporter =
                 new BrokerLoadDataReporter(pulsar, brokerRegistry.getBrokerId(), brokerLoadDataStore);
 
@@ -216,10 +220,12 @@ public void start() throws PulsarServerException {
                         interval,
                         interval, TimeUnit.MILLISECONDS);
 
-        // TODO: Start bundle split scheduler.
         this.unloadScheduler = new UnloadScheduler(
                 pulsar.getLoadManagerExecutor(), unloadManager, context, serviceUnitStateChannel);
         this.unloadScheduler.start();
+        this.splitScheduler = new SplitScheduler(
+                pulsar, serviceUnitStateChannel, splitManager, splitCounter, splitMetrics, context);
+        this.splitScheduler.start();
         this.started = true;
     }
 
@@ -380,6 +386,7 @@ public void close() throws PulsarServerException {
             this.brokerLoadDataStore.close();
             this.topBundlesLoadDataStore.close();
             this.unloadScheduler.close();
+            this.splitScheduler.close();
         } catch (IOException ex) {
             throw new PulsarServerException(ex);
         } finally {
@@ -411,13 +418,6 @@ private void updateUnloadMetrics(UnloadDecision decision) {
         this.unloadMetrics.set(unloadCounter.toMetrics(pulsar.getAdvertisedAddress()));
     }
 
-    private void updateSplitMetrics(List<SplitDecision> decisions) {
-        for (var decision : decisions) {
-            splitCounter.update(decision);
-        }
-        this.splitMetrics.set(splitCounter.toMetrics(pulsar.getAdvertisedAddress()));
-    }
-
     public List<Metrics> getMetrics() {
         List<Metrics> metricsCollection = new ArrayList<>();
 

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

@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.loadbalance.extensions.manager;
+
+import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Failure;
+import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Unknown;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitState;
+import org.apache.pulsar.broker.loadbalance.extensions.channel.ServiceUnitStateData;
+import org.apache.pulsar.broker.loadbalance.extensions.models.SplitCounter;
+import org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision;
+
+/**
+ * Split manager.
+ */
+@Slf4j
+public class SplitManager implements StateChangeListener {
+
+
+    private final Map<String, CompletableFuture<Void>> inFlightSplitRequests;
+
+    private final SplitCounter counter;
+
+    public SplitManager(SplitCounter splitCounter) {
+        this.inFlightSplitRequests = new ConcurrentHashMap<>();
+        this.counter = splitCounter;
+    }
+
+    private void complete(String serviceUnit, Throwable ex) {
+        inFlightSplitRequests.computeIfPresent(serviceUnit, (__, future) -> {
+            if (!future.isDone()) {
+                if (ex != null) {
+                    future.completeExceptionally(ex);
+                } else {
+                    future.complete(null);
+                }
+            }
+            return null;
+        });
+    }
+
+    public CompletableFuture<Void> waitAsync(CompletableFuture<Void> eventPubFuture,
+                                             String bundle,
+                                             SplitDecision decision,
+                                             long timeout,
+                                             TimeUnit timeoutUnit) {
+        return eventPubFuture
+                .thenCompose(__ -> inFlightSplitRequests.computeIfAbsent(bundle, ignore -> {
+                    log.info("Published the bundle split event for bundle:{}. "
+                                    + "Waiting the split event to complete. Timeout: {} {}",
+                            bundle, timeout, timeoutUnit);
+                    CompletableFuture<Void> future = new CompletableFuture<>();
+                    future.orTimeout(timeout, timeoutUnit).whenComplete((v, ex) -> {
+                        if (ex != null) {
+                            inFlightSplitRequests.remove(bundle);
+                            log.warn("Timed out while waiting for the bundle split event: {}", bundle, ex);
+                        }
+                    });
+                    return future;
+                }))
+                .whenComplete((__, ex) -> {
+                    if (ex != null) {
+                        log.error("Failed the bundle split event for bundle:{}", bundle, ex);
+                        counter.update(Failure, Unknown);
+                    } else {
+                        log.info("Completed the bundle split event for bundle:{}", bundle);
+                        counter.update(decision);
+                    }
+                });
+    }
+
+    @Override
+    public void handleEvent(String serviceUnit, ServiceUnitStateData data, Throwable t) {
+        ServiceUnitState state = ServiceUnitStateData.state(data);
+        if (t != null && inFlightSplitRequests.containsKey(serviceUnit)) {
+            this.complete(serviceUnit, t);
+            return;
+        }
+        switch (state) {
+            case Deleted, Owned, Init -> this.complete(serviceUnit, t);
+            default -> {
+                if (log.isDebugEnabled()) {
+                    log.debug("Handling {} for service unit {}", data, serviceUnit);
+                }
+            }
+        }
+    }
+
+    public void close() {
+        inFlightSplitRequests.forEach((bundle, future) -> {
+            if (!future.isDone()) {
+                String msg = String.format("Splitting bundle: %s, but the manager already closed.", bundle);
+                log.warn(msg);
+                future.completeExceptionally(new IllegalStateException(msg));
+            }
+        });
+        inFlightSplitRequests.clear();
+    }
+}

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

@@ -19,10 +19,8 @@
 package org.apache.pulsar.broker.loadbalance.extensions.models;
 
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Failure;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Skip;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Success;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Admin;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Balanced;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Bandwidth;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.MsgRate;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Sessions;
@@ -32,39 +30,45 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import org.apache.commons.lang3.mutable.MutableLong;
+import java.util.concurrent.atomic.AtomicLong;
 import org.apache.pulsar.common.stats.Metrics;
 
 /**
  * Defines the information required for a service unit split(e.g. bundle split).
  */
 public class SplitCounter {
 
-    long splitCount = 0;
-
-    final Map<SplitDecision.Label, Map<SplitDecision.Reason, MutableLong>> breakdownCounters;
+    private long splitCount = 0;
+    private final Map<SplitDecision.Label, Map<SplitDecision.Reason, AtomicLong>> breakdownCounters;
+    private volatile long updatedAt = 0;
 
     public SplitCounter() {
         breakdownCounters = Map.of(
                 Success, Map.of(
-                        Topics, new MutableLong(),
-                        Sessions, new MutableLong(),
-                        MsgRate, new MutableLong(),
-                        Bandwidth, new MutableLong(),
-                        Admin, new MutableLong()),
-                Skip, Map.of(
-                        Balanced, new MutableLong()
-                        ),
+                        Topics, new AtomicLong(),
+                        Sessions, new AtomicLong(),
+                        MsgRate, new AtomicLong(),
+                        Bandwidth, new AtomicLong(),
+                        Admin, new AtomicLong()),
                 Failure, Map.of(
-                        Unknown, new MutableLong())
+                        Unknown, new AtomicLong())
         );
     }
 
     public void update(SplitDecision decision) {
         if (decision.label == Success) {
             splitCount++;
         }
-        breakdownCounters.get(decision.getLabel()).get(decision.getReason()).increment();
+        breakdownCounters.get(decision.getLabel()).get(decision.getReason()).incrementAndGet();
+        updatedAt = System.currentTimeMillis();
+    }
+
+    public void update(SplitDecision.Label label, SplitDecision.Reason reason) {
+        if (label == Success) {
+            splitCount++;
+        }
+        breakdownCounters.get(label).get(reason).incrementAndGet();
+        updatedAt = System.currentTimeMillis();
     }
 
     public List<Metrics> toMetrics(String advertisedBrokerAddress) {
@@ -77,22 +81,26 @@ public List<Metrics> toMetrics(String advertisedBrokerAddress) {
         m.put("brk_lb_bundles_split_total", splitCount);
         metrics.add(m);
 
-        for (Map.Entry<SplitDecision.Label, Map<SplitDecision.Reason, MutableLong>> etr
+
+        for (Map.Entry<SplitDecision.Label, Map<SplitDecision.Reason, AtomicLong>> etr
                 : breakdownCounters.entrySet()) {
             var result = etr.getKey();
-            for (Map.Entry<SplitDecision.Reason, MutableLong> counter : etr.getValue().entrySet()) {
+            for (Map.Entry<SplitDecision.Reason, AtomicLong> counter : etr.getValue().entrySet()) {
                 var reason = counter.getKey();
                 var count = counter.getValue();
                 Map<String, String> breakdownDims = new HashMap<>(dimensions);
                 breakdownDims.put("result", result.toString());
                 breakdownDims.put("reason", reason.toString());
                 Metrics breakdownMetric = Metrics.create(breakdownDims);
-                breakdownMetric.put("brk_lb_bundles_split_breakdown_total", count);
+                breakdownMetric.put("brk_lb_bundles_split_breakdown_total", count.get());
                 metrics.add(breakdownMetric);
             }
         }
 
         return metrics;
     }
 
+    public long updatedAt() {
+        return updatedAt;
+    }
 }

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

@@ -19,9 +19,7 @@
 package org.apache.pulsar.broker.loadbalance.extensions.models;
 
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Failure;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Skip;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Label.Success;
-import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Balanced;
 import static org.apache.pulsar.broker.loadbalance.extensions.models.SplitDecision.Reason.Unknown;
 import lombok.Data;
 
@@ -36,7 +34,6 @@ public class SplitDecision {
 
     public enum Label {
         Success,
-        Skip,
         Failure
     }
 
@@ -46,7 +43,6 @@ public enum Reason {
         MsgRate,
         Bandwidth,
         Admin,
-        Balanced,
         Unknown
     }
 
@@ -62,11 +58,6 @@ public void clear() {
         reason = null;
     }
 
-    public void skip() {
-        label = Skip;
-        reason = Balanced;
-    }
-
     public void succeed(Reason reason) {
         label = Success;
         this.reason = reason;