[server] Add off-loop coordinator health cache (mechanism only)

fluss-server/src/main/java/org/apache/fluss/server/coordinator/ClusterHealthSnapshot.java

@@ -0,0 +1,80 @@
+/*
+ * 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.fluss.server.coordinator;
+
+import java.util.Collections;
+import java.util.Map;
+
+/**
+ * An immutable, point-in-time view of cluster-wide and per-tablet-server replica/leader health,
+ * derived from {@link CoordinatorContext}.
+ *
+ * <p>{@code numReplicas}/{@code inSyncReplicas}/{@code numLeaderReplicas}/{@code
+ * activeLeaderReplicas} are the same cluster-wide aggregates {@code
+ * CoordinatorService#computeClusterHealth} reports; {@link #tabletServerLoads()} additionally
+ * attributes replicas/ISR membership/leadership to the specific server holding them. Both are
+ * computed together from a single pass over {@link CoordinatorContext#getAllBuckets()}.
+ *
+ * <p>Instances are published by {@link CoordinatorHealthCache} and are safe to read from any thread
+ * without synchronization: once constructed, an instance is never mutated.
+ */
+public final class ClusterHealthSnapshot {
+
+    public static final ClusterHealthSnapshot EMPTY =
+            new ClusterHealthSnapshot(0, 0, 0, 0, Collections.emptyMap());
+
+    private final int numReplicas;
+    private final int inSyncReplicas;
+    private final int numLeaderReplicas;
+    private final int activeLeaderReplicas;
+    private final Map<Integer, TabletServerLoad> tabletServerLoads;
+
+    ClusterHealthSnapshot(
+            int numReplicas,
+            int inSyncReplicas,
+            int numLeaderReplicas,
+            int activeLeaderReplicas,
+            Map<Integer, TabletServerLoad> tabletServerLoads) {
+        this.numReplicas = numReplicas;
+        this.inSyncReplicas = inSyncReplicas;
+        this.numLeaderReplicas = numLeaderReplicas;
+        this.activeLeaderReplicas = activeLeaderReplicas;
+        this.tabletServerLoads = Collections.unmodifiableMap(tabletServerLoads);
+    }
+
+    public int numReplicas() {
+        return numReplicas;
+    }
+
+    public int inSyncReplicas() {
+        return inSyncReplicas;
+    }
+
+    public int numLeaderReplicas() {
+        return numLeaderReplicas;
+    }
+
+    public int activeLeaderReplicas() {
+        return activeLeaderReplicas;
+    }
+
+    /** Per-tablet-server load, keyed by server id. Includes live and shutting-down servers. */
+    public Map<Integer, TabletServerLoad> tabletServerLoads() {
+        return tabletServerLoads;
+    }
+}

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -188,6 +188,7 @@ public class CoordinatorEventProcessor implements EventProcessor {
     private final CoordinatorChangeWatcher coordinatorChangeWatcher;
     private final TabletServerChangeWatcher tabletServerChangeWatcher;
     private final CoordinatorMetadataCache serverMetadataCache;
+    private final CoordinatorHealthCache healthCache;
     private final CoordinatorRequestBatch coordinatorRequestBatch;
     private final String internalListenerName;
     private final CoordinatorMetricGroup coordinatorMetricGroup;
@@ -218,7 +219,10 @@ public CoordinatorEventProcessor(
         this.coordinatorChannelManager = coordinatorChannelManager;
         this.coordinatorContext = coordinatorContext;
         this.replicaCapacityController = replicaCapacityController;
-        this.coordinatorEventManager = new CoordinatorEventManager(this, coordinatorMetricGroup);
+        this.healthCache = new CoordinatorHealthCache();
+        this.coordinatorEventManager =
+                new CoordinatorEventManager(
+                        this, coordinatorContext, healthCache, coordinatorMetricGroup);
         this.replicaStateMachine =
                 new ReplicaStateMachine(
                         coordinatorContext,
@@ -300,6 +304,10 @@ public CoordinatorContext getCoordinatorContext() {
         return coordinatorContext;
     }
 
+    public CoordinatorHealthCache getHealthCache() {
+        return healthCache;
+    }
+
     @VisibleForTesting
     TableLifecycleThrottler getLifecycleThrottler() {
         return lifecycleThrottler;
@@ -501,6 +509,11 @@ private void initCoordinatorContext() throws Exception {
                 tabletServerInfoList,
                 coordinatorContext.getServerTags());
         updateTabletServerMetadataCacheWhenStartup(tabletServerInfoList);
+        // Warm the health cache once, unconditionally, so it isn't empty before the first
+        // relevant mutation lands. Bulk-loading buckets above does not mark it dirty on purpose
+        // (that would mean one dirty-mark per bucket during startup, for no benefit); this single
+        // explicit refresh covers it instead.
+        healthCache.refresh(coordinatorContext, true);
 
         // Auto-partition initialization schedules creation checks immediately. Start it only after
         // the observed KV leader replica count and live tablet server resources are restored.
@@ -669,12 +682,16 @@ private void enqueueRetryOfflineLeaderEventSafely() {
     public void process(CoordinatorEvent event) {
         if (event instanceof CreateTableEvent) {
             processCreateTable((CreateTableEvent) event);
+            healthCache.onTopologyChanged();
         } else if (event instanceof CreatePartitionEvent) {
             processCreatePartition((CreatePartitionEvent) event);
+            healthCache.onTopologyChanged();
         } else if (event instanceof DropTableEvent) {
             processDropTable((DropTableEvent) event);
+            healthCache.onTopologyChanged();
         } else if (event instanceof DropPartitionEvent) {
             processDropPartition((DropPartitionEvent) event);
+            healthCache.onTopologyChanged();
         } else if (event instanceof SchemaChangeEvent) {
             SchemaChangeEvent schemaChangeEvent = (SchemaChangeEvent) event;
             processSchemaChange(schemaChangeEvent);
@@ -693,8 +710,10 @@ public void process(CoordinatorEvent event) {
             processDeadCoordinator((DeadCoordinatorEvent) event);
         } else if (event instanceof NewTabletServerEvent) {
             processNewTabletServer((NewTabletServerEvent) event);
+            healthCache.onTabletServerRegistered();
         } else if (event instanceof DeadTabletServerEvent) {
             processDeadTabletServer((DeadTabletServerEvent) event);
+            healthCache.onTabletServerDied();
         } else if (event instanceof AdjustIsrReceivedEvent) {
             AdjustIsrReceivedEvent adjustIsrReceivedEvent = (AdjustIsrReceivedEvent) event;
             CompletableFuture<AdjustIsrResponse> callback =
@@ -733,12 +752,21 @@ public void process(CoordinatorEvent event) {
             AddServerTagEvent addServerTagEvent = (AddServerTagEvent) event;
             completeFromCallable(
                     addServerTagEvent.getRespCallback(),
-                    () -> processAddServerTag(addServerTagEvent));
+                    () -> {
+                        AddServerTagResponse response = processAddServerTag(addServerTagEvent);
+                        healthCache.onTopologyChanged();
+                        return response;
+                    });
         } else if (event instanceof RemoveServerTagEvent) {
             RemoveServerTagEvent removeServerTagEvent = (RemoveServerTagEvent) event;
             completeFromCallable(
                     removeServerTagEvent.getRespCallback(),
-                    () -> processRemoveServerTag(removeServerTagEvent));
+                    () -> {
+                        RemoveServerTagResponse response =
+                                processRemoveServerTag(removeServerTagEvent);
+                        healthCache.onTopologyChanged();
+                        return response;
+                    });
         } else if (event instanceof RebalanceEvent) {
             RebalanceEvent rebalanceEvent = (RebalanceEvent) event;
             completeFromCallable(
@@ -1790,6 +1818,7 @@ private void onBucketReassignment(
 
             // A2. Set RS = TRS, AR = [], RR = [] in memory.
             coordinatorContext.updateBucketReplicaAssignment(tableBucket, reassignment.replicas);
+            healthCache.onTopologyChanged();
             updateReplicaAssignmentForBucket(tableBucket, reassignment.replicas);
 
             // A3. replicas in AR -> NewReplica
@@ -1815,6 +1844,7 @@ private void onBucketReassignment(
             maybeReassignedBucketLeaderIfRequired(tableBucket, targetReplicas);
             // B3. Set RS = TRS, AR = [], RR = [] in memory.
             coordinatorContext.updateBucketReplicaAssignment(tableBucket, targetReplicas);
+            healthCache.onTopologyChanged();
             // B4. Re-send LeaderAndIsr request with new leader and a new RS (using TRS) and same
             // isr to every tabletServer in TRS.
             updateBucketEpochAndSendRequest(tableBucket, targetReplicas);
@@ -1991,7 +2021,14 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
         }
 
         // update coordinator leader and isr cache.
-        newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
+        newLeaderAndIsrList.forEach(
+                (tableBucket, newLeaderAndIsr) -> {
+                    coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr);
+                    healthCache.onBucketLeaderAndIsrChanged(
+                            tableBucket,
+                            coordinatorContext.getAssignment(tableBucket),
+                            Optional.of(newLeaderAndIsr));
+                });
 
         // First, try to judge whether the bucket is in rebalance task when isr change.
         newLeaderAndIsrList.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange);
@@ -2552,6 +2589,10 @@ private void updateBucketEpochAndSendRequest(TableBucket tableBucket, List<Integ
         LeaderAndIsr newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(leaderAndIsr.isr());
 
         coordinatorContext.putBucketLeaderAndIsr(tableBucket, newLeaderAndIsr);
+        healthCache.onBucketLeaderAndIsrChanged(
+                tableBucket,
+                coordinatorContext.getAssignment(tableBucket),
+                Optional.of(newLeaderAndIsr));
         zooKeeperClient.updateLeaderAndIsr(
                 tableBucket, newLeaderAndIsr, coordinatorContext.getCoordinatorZkVersion());