[helm] First-class coordinator HA: standby-aware readiness gate

fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java

@@ -787,13 +787,22 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
      *   <li>{@link ClusterHealthStatus#RED} — one or more leader replicas have not yet been
      *       confirmed active (e.g., leader election or KV snapshot recovery is still in progress).
      *   <li>{@link ClusterHealthStatus#UNKNOWN} — the Coordinator was unable to determine cluster
-     *       health (e.g., the server does not support this API).
+     *       health (e.g., the server does not support this API), or the answering coordinator is a
+     *       standby (see below).
      * </ul>
      *
      * <p>This API is designed for the situation like a Kubernetes readiness-probe gate during
      * rolling upgrades: only proceed to the next pod when the status is {@code GREEN}, ensuring all
      * replicas have fully recovered before the next server is restarted.
      *
+     * <p>Unlike other admin operations, a <b>standby</b> coordinator answers this request instead
+     * of rejecting it with a {@code NotCoordinatorLeaderException} (so a readiness probe can gate
+     * on it). If the client's coordinator address is stale — e.g. during a coordinator failover,
+     * before the client's metadata refreshes — the returned {@link ClusterHealth} carries {@link
+     * ClusterHealth#isServedByLeader()} {@code false}, status {@code UNKNOWN}, and zeroed replica
+     * counts. Callers that monitor cluster health should check {@link
+     * ClusterHealth#isServedByLeader()} before interpreting the counts.
+     *
      * @return a {@link CompletableFuture} that completes with the cluster health information.
      * @since 1.0
      */

fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealth.java

@@ -24,6 +24,11 @@
 /**
  * Cluster health information returned by {@link Admin#getClusterHealth()}.
  *
+ * <p>A standby coordinator answers this request instead of rejecting it (so Kubernetes readiness
+ * probes can gate on it). A response served by a standby carries {@link #isServedByLeader()} {@code
+ * false}, status {@link ClusterHealthStatus#UNKNOWN}, and zeroed replica counts — callers that
+ * monitor cluster health should check {@link #isServedByLeader()} before interpreting the counts.
+ *
  * @since 1.0
  */
 @PublicEvolving
@@ -34,18 +39,24 @@ public final class ClusterHealth {
     private final int numLeaderReplicas;
     private final int activeLeaderReplicas;
     private final ClusterHealthStatus status;
+    private final boolean servedByLeader;
+    private final boolean leaderElected;
 
     public ClusterHealth(
             int numReplicas,
             int inSyncReplicas,
             int numLeaderReplicas,
             int activeLeaderReplicas,
-            ClusterHealthStatus status) {
+            ClusterHealthStatus status,
+            boolean servedByLeader,
+            boolean leaderElected) {
         this.numReplicas = numReplicas;
         this.inSyncReplicas = inSyncReplicas;
         this.numLeaderReplicas = numLeaderReplicas;
         this.activeLeaderReplicas = activeLeaderReplicas;
         this.status = Objects.requireNonNull(status, "status");
+        this.servedByLeader = servedByLeader;
+        this.leaderElected = leaderElected;
     }
 
     public int getNumReplicas() {
@@ -68,6 +79,26 @@ public ClusterHealthStatus getStatus() {
         return status;
     }
 
+    /**
+     * Whether the coordinator that answered is the current leader. {@code false} means a standby
+     * answered (e.g. the client's coordinator address was stale during a failover): the status is
+     * {@link ClusterHealthStatus#UNKNOWN} and the replica counts are zero, not cluster facts.
+     * Responses from servers that predate this field report {@code true} — a standby of those
+     * versions rejects the request instead of answering.
+     */
+    public boolean isServedByLeader() {
+        return servedByLeader;
+    }
+
+    /**
+     * Whether the coordinator group currently has an elected leader — the answering server or any
+     * other participant. Only meaningful when {@link #isServedByLeader()} is {@code false}; a
+     * leader-served response always reports {@code true}.
+     */
+    public boolean isLeaderElected() {
+        return leaderElected;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -81,13 +112,21 @@ public boolean equals(Object o) {
                 && inSyncReplicas == that.inSyncReplicas
                 && numLeaderReplicas == that.numLeaderReplicas
                 && activeLeaderReplicas == that.activeLeaderReplicas
-                && status == that.status;
+                && status == that.status
+                && servedByLeader == that.servedByLeader
+                && leaderElected == that.leaderElected;
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
-                numReplicas, inSyncReplicas, numLeaderReplicas, activeLeaderReplicas, status);
+                numReplicas,
+                inSyncReplicas,
+                numLeaderReplicas,
+                activeLeaderReplicas,
+                status,
+                servedByLeader,
+                leaderElected);
     }
 
     @Override
@@ -103,6 +142,10 @@ public String toString() {
                 + activeLeaderReplicas
                 + ", status="
                 + status
+                + ", servedByLeader="
+                + servedByLeader
+                + ", leaderElected="
+                + leaderElected
                 + '}';
     }
 }

fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java

@@ -882,12 +882,18 @@ public static GetTableStatsRequest makeGetTableStatsRequest(List<TableBucket> bu
     }
 
     public static ClusterHealth toClusterHealth(GetClusterHealthResponse resp) {
+        // A response without is_leader comes from a leader that predates the field — a standby
+        // of those versions rejects the request instead of answering.
+        boolean servedByLeader = !resp.hasIsLeader() || resp.isIsLeader();
+        boolean leaderElected = resp.hasLeaderElected() ? resp.isLeaderElected() : servedByLeader;
         return new ClusterHealth(
                 resp.getNumReplicas(),
                 resp.getInSyncReplicas(),
                 resp.getNumLeaderReplicas(),
                 resp.getActiveLeaderReplicas(),
-                toClusterHealthStatus(resp.getStatus()));
+                toClusterHealthStatus(resp.getStatus()),
+                servedByLeader,
+                leaderElected);
     }
 
     private static ClusterHealthStatus toClusterHealthStatus(int pbStatus) {

fluss-dist/src/main/resources/bin/readiness-check.sh

@@ -24,13 +24,17 @@
 #
 # Two-step readiness probe for Kubernetes StatefulSet rolling upgrades:
 #
-#   Step 1 — Local TCP port check: verify this TabletServer process is alive
+#   Step 1 — Local TCP port check: verify this server process is alive
 #            and has bound its RPC port. Fast, no external dependency.
 #
-#   Step 2 — Cluster health check: call the LOCAL TabletServer's Cluster
-#            Health API (which forwards to the Coordinator over the internal
-#            listener) and pass only if status is GREEN.
-#            YELLOW/RED/UNKNOWN means recovery is incomplete — block the upgrade.
+#   Step 2 — Cluster health check, role-dependent (READINESS_ROLE):
+#            tablet (default) — call the LOCAL TabletServer's Cluster Health
+#            API (which forwards to the Coordinator over the internal
+#            listener) and pass only if status is GREEN. YELLOW/RED/UNKNOWN
+#            means recovery is incomplete — block the upgrade.
+#            coordinator — call the LOCAL CoordinatorServer's Cluster Health
+#            API. A leader passes regardless of cluster color; a standby
+#            passes only when the coordinator group has an elected leader.
 #
 # Both steps must pass for the pod to be marked Ready.
 #
@@ -41,6 +45,8 @@
 #
 # Environment variables (set by helm template or container spec):
 #   FLUSS_HOME                    - Fluss installation directory
+#   READINESS_ROLE                - Role of the local server the probe talks to:
+#       "tablet" (default) or "coordinator".
 #   READINESS_TIMEOUT_MS          - Timeout for Health API call (default: 5000)
 #   READINESS_TCP_HOST            - TabletServer host the probe talks to. Defaults
 #       to ${POD_IP} when set (the typical Kubernetes case where bind.listeners
@@ -66,6 +72,7 @@ set -o pipefail
 # ---- Configuration ----
 
 FLUSS_HOME="${FLUSS_HOME:-/opt/fluss}"
+ROLE="${READINESS_ROLE:-tablet}"
 TIMEOUT_MS="${READINESS_TIMEOUT_MS:-5000}"
 # Probe the local sidecar tablet inside the same pod — it forwards the request
 # to the Coordinator internally, so the probe never has to know the
@@ -145,6 +152,7 @@ run_recovery_check() {
         -Xmx64m \
         -classpath "${classpath}" \
         org.apache.fluss.server.tools.ClusterHealthReadinessCheck \
+        --role "${ROLE}" \
         --host "${TCP_HOST}" \
         --port "${TCP_PORT}" \
         --timeoutMs "${timeout_ms}" 2>&1)

fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java

@@ -65,7 +65,12 @@ public void processRequest(FlussRequest request) {
                             request.getPrincipal()));
             // check if the coordinator server is the current leader if the API is a coordinator
             // TODO: we should only check coordinator APIs instead of all APIs
-            if (isCoordinator && api.getApiKey() != ApiKeys.API_VERSIONS) {
+            // GET_CLUSTER_HEALTH is answerable by a standby coordinator: it reports the
+            // server's own role and election state so Kubernetes readiness probes can tell
+            // a healthy standby from a wedged one.
+            if (isCoordinator
+                    && api.getApiKey() != ApiKeys.API_VERSIONS
+                    && api.getApiKey() != ApiKeys.GET_CLUSTER_HEALTH) {
                 if (!((CoordinatorGateway) service).isLeader()) {
                     request.fail(
                             new NotCoordinatorLeaderException(

fluss-rpc/src/main/proto/FlussApi.proto

@@ -814,6 +814,12 @@ message GetClusterHealthResponse {
   required int32 num_leader_replicas = 3;
   required int32 active_leader_replicas = 4;
   required int32 status = 5; // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
+  // Whether the answering coordinator is the current leader. A standby coordinator
+  // answers with is_leader=false, status=UNKNOWN and zeroed replica counts.
+  optional bool is_leader = 6;
+  // Whether the coordinator group currently has an elected leader (possibly another
+  // server). Lets a standby's readiness probe certify a functioning group.
+  optional bool leader_elected = 7;
 }
 
 

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

@@ -190,6 +190,35 @@ public boolean isLeader() {
         return !closing.get() && state == State.LEADER;
     }
 
+    /**
+     * Returns whether the coordinator group currently has an elected leader — this server or any
+     * other participant. Returns {@code false} when the state cannot be determined (e.g. ZooKeeper
+     * unreachable), because an unknown leader must not be reported as present to a readiness probe.
+     *
+     * <p>On a standby this performs a synchronous ZooKeeper read, so it is intended for
+     * probe-frequency callers only (the Kubernetes readiness probe calls it at most every few
+     * seconds), not for hot paths.
+     *
+     * <p>TODO: when leader initialization fails, {@code becomeLeader} transitions this server back
+     * to STANDBY but the {@link LeaderLatch} keeps the ZK leadership, so this method can report an
+     * elected leader while no functional leader exists until the latch is released or the session
+     * expires. Pre-existing election behavior; revisit together with latch relinquishing.
+     */
+    public boolean isLeaderElected() {
+        if (closing.get()) {
+            return false;
+        }
+        if (state == State.LEADER) {
+            return true;
+        }
+        try {
+            return leaderLatch.getLeader().isLeader();
+        } catch (Exception e) {
+            LOG.debug("Failed to read leader election state for server {}", serverId, e);
+            return false;
+        }
+    }
+
     private void submitLeadershipEvent(Runnable leadershipEvent) {
         if (closing.get()) {
             return;

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

@@ -1584,12 +1584,32 @@ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
             authorizer.authorize(currentSession(), OperationType.DESCRIBE, Resource.cluster());
         }
 
+        if (!isLeader()) {
+            // A standby has no CoordinatorContext (the event processor is leader-only), so it
+            // answers from its own election state: status UNKNOWN, zeroed counts, and the role
+            // fields a readiness probe needs to tell a healthy standby from a wedged one.
+            return CompletableFuture.completedFuture(computeStandbyClusterHealth());
+        }
+
         AccessContextEvent<GetClusterHealthResponse> event =
-                new AccessContextEvent<>(CoordinatorService::computeClusterHealth);
+                new AccessContextEvent<>(
+                        ctx -> computeClusterHealth(ctx).setIsLeader(true).setLeaderElected(true));
         eventManagerSupplier.get().put(event);
         return event.getResultFuture();
     }
 
+    private GetClusterHealthResponse computeStandbyClusterHealth() {
+        GetClusterHealthResponse response = new GetClusterHealthResponse();
+        response.setNumReplicas(0);
+        response.setInSyncReplicas(0);
+        response.setNumLeaderReplicas(0);
+        response.setActiveLeaderReplicas(0);
+        response.setStatus(3); // PbClusterHealthStatus.UNKNOWN
+        response.setIsLeader(false);
+        response.setLeaderElected(coordinatorLeaderElection.isLeaderElected());
+        return response;
+    }
+
     @VisibleForTesting
     static GetClusterHealthResponse computeClusterHealth(CoordinatorContext ctx) {
         GetClusterHealthResponse response = new GetClusterHealthResponse();