[improve][broker] PIP-192 Added Deleted and Init states in ServiceUnitState

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

@@ -2450,6 +2450,7 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private long namespaceBundleUnloadingTimeoutMs = 60000;
 
+    /**** --- Load Balancer Extension. --- ****/
     @FieldContext(
             category = CATEGORY_LOAD_BALANCER,
             dynamic = true,
@@ -2525,6 +2526,22 @@ The delayed message index bucket time step(in seconds) in per bucket snapshot se
     )
     private double loadBalancerBundleLoadReportPercentage = 10;
 
+    @FieldContext(
+            category = CATEGORY_LOAD_BALANCER,
+            doc = "After this delay, the service-unit state channel tombstones any service units (e.g., bundles) "
+                    + "in semi-terminal states. For example, after splits, parent bundles will be `Disabled`, "
+                    + "and then after this delay, the parent bundles' state will be removed(tombstoned) "
+                    + "in the service-unit state channel. "
+                    + "Pulsar does not immediately remove such semi-terminal states "
+                    + "to avoid unnecessary system confusion, "
+                    + "as the bundles in the `Tombstoned` state might temporarily look available to reassign. "
+                    + "Rarely, one could lower this delay in order to aggressively clean "
+                    + "the service-unit state channel when there are a large number of bundles. "
+                    + "minimum value = 30 secs"
+                    + "(only used in load balancer extension logics)"
+    )
+    private long loadBalancerServiceUnitStateCleanUpDelayTimeInSeconds = 604800;
+
     /**** --- Replication. --- ****/
     @FieldContext(
         category = CATEGORY_REPLICATION,

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

@@ -142,4 +142,8 @@ public void doNamespaceBundleSplit() {
         throw new UnsupportedOperationException();
     }
 
+    public ExtensibleLoadManagerImpl get() {
+        return loadManager;
+    }
+
 }

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

@@ -24,55 +24,34 @@
 /**
  * Defines the possible states for service units.
  *
- * The following diagram defines the valid state changes
- *
- *                  ┌───────────┐
- *       ┌──────────┤ released  │◄────────┐
- *       │own       └───────────┘         │release
- *       │                                │
- *       │                                │
- *       ▼                                │
- *    ┌────────┐  assign(transfer)  ┌─────┴────┐
- *    │        ├───────────────────►│          │
- *    │ owned  │                    │ assigned │
- *    │        │◄───────────────────┤          │
- *    └──┬─────┤      own           └──────────┘
- *       │  ▲  │                         ▲
- *       │  │  │                         │
- *       │  │  └──────────────┐          │
- *       │  │                 │          │
- *       │  │        unload   │          │ assign(assignment)
- * split │  │                 │          │
- *       │  │                 │          │
- *       │  │ create(child)   │          │
- *       │  │                 │          │
- *       ▼  │                 │          │
- *    ┌─────┴─────┐           └─────►┌───┴──────┐
- *    │           │                  │          │
- *    │ splitting ├────────────────► │   free   │
- *    │           │   discard(parent)│          │
- *    └───────────┘                  └──────────┘
+ * Refer to Service Unit State Channel in https://github.com/apache/pulsar/issues/16691 for additional details.
  */
 public enum ServiceUnitState {
 
-    Free, // not owned by any broker (terminal state)
+    Init, // initializing the state. no previous state(terminal state)
+
+    Free, // not owned by any broker (semi-terminal state)
 
     Owned, // owned by a broker (terminal state)
 
     Assigned, // the ownership is assigned(but the assigned broker has not been notified the ownership yet)
 
     Released, // the source broker's ownership has been released (e.g. the topic connections are closed)
 
-    Splitting; // the service unit(e.g. bundle) is in the process of splitting.
+    Splitting, // the service unit(e.g. bundle) is in the process of splitting.
+
+    Disabled; // disabled in the system (semi-terminal state)
 
     private static Map<ServiceUnitState, Set<ServiceUnitState>> validTransitions = Map.of(
-            // (Free -> Released | Splitting) transitions are required
-            // when the topic is compacted in the middle of transfer or split.
-            Free, Set.of(Owned, Assigned, Released, Splitting),
-            Owned, Set.of(Assigned, Splitting, Free),
-            Assigned, Set.of(Owned, Released, Free),
-            Released, Set.of(Owned, Free),
-            Splitting, Set.of(Free)
+            // (Init -> all states) transitions are required
+            // when the topic is compacted in the middle of assign, transfer or split.
+            Init, Set.of(Free, Owned, Assigned, Released, Splitting, Disabled, Init),
+            Free, Set.of(Assigned, Init),
+            Owned, Set.of(Assigned, Splitting, Free, Init),
+            Assigned, Set.of(Owned, Released, Init),
+            Released, Set.of(Owned, Init),
+            Splitting, Set.of(Disabled, Init),
+            Disabled, Set.of(Init)
     );
 
     public static boolean isValidTransition(ServiceUnitState from, ServiceUnitState to) {