diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java index 5d749b6d432..e34f6de696f 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/Admin.java @@ -787,13 +787,22 @@ CompletableFuture registerProducerOffsets( *
  • {@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). *
  • {@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). * * *

    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. * + *

    Unlike other admin operations, a standby 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 */ diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealth.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealth.java index 8f751bc1691..266b2512215 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealth.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealth.java @@ -24,6 +24,11 @@ /** * Cluster health information returned by {@link Admin#getClusterHealth()}. * + *

    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 + '}'; } } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java index 3c7512945dc..24d1fb8630e 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java @@ -882,12 +882,18 @@ public static GetTableStatsRequest makeGetTableStatsRequest(List 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) { diff --git a/fluss-dist/src/main/resources/bin/readiness-check.sh b/fluss-dist/src/main/resources/bin/readiness-check.sh index 7d2ac5d9a3f..1e4a4ea21f1 100644 --- a/fluss-dist/src/main/resources/bin/readiness-check.sh +++ b/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) diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java index 2ff5e0691bb..d3abc8bf4e9 100644 --- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/netty/server/FlussRequestHandler.java +++ b/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( diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index d03acb92fbe..4baf0224874 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/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; } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java index 0ba51dafe09..c913b995f89 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorLeaderElection.java +++ b/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. + * + *

    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. + * + *

    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; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java index 04119bb0fd9..04b674e8680 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java @@ -1584,12 +1584,32 @@ public CompletableFuture 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 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(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/tools/ClusterHealthReadinessCheck.java b/fluss-server/src/main/java/org/apache/fluss/server/tools/ClusterHealthReadinessCheck.java index 6e9c70ae13d..65d1794cabb 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/tools/ClusterHealthReadinessCheck.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/tools/ClusterHealthReadinessCheck.java @@ -21,10 +21,13 @@ import org.apache.fluss.cluster.ServerType; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.NotCoordinatorLeaderException; import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.metrics.registry.MetricRegistryImpl; import org.apache.fluss.rpc.GatewayClientProxy; import org.apache.fluss.rpc.RpcClient; +import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway; +import org.apache.fluss.rpc.gateway.CoordinatorGateway; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.GetClusterHealthRequest; import org.apache.fluss.rpc.messages.GetClusterHealthResponse; @@ -32,18 +35,25 @@ import org.apache.fluss.utils.ExceptionUtils; import java.util.Collections; +import java.util.Locale; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; /** - * Lightweight readiness-check CLI for Fluss tablet-server pods. + * Lightweight readiness-check CLI for Fluss server pods. * - *

    The probe connects to a tablet server (typically the local one on {@code 127.0.0.1}) and - * issues a single {@code getClusterHealth} call. The tablet server forwards the request to the - * coordinator over its internal listener and returns the cluster-wide health snapshot. This means - * the readiness probe never has to know the coordinator's address and survives coordinator pod - * restarts cleanly. + *

    In the default {@code tablet} role, the probe connects to a tablet server (typically the local + * one on {@code 127.0.0.1}) and issues a single {@code getClusterHealth} call. The tablet server + * forwards the request to the coordinator over its internal listener and returns the cluster-wide + * health snapshot. This means the readiness probe never has to know the coordinator's address and + * survives coordinator pod restarts cleanly. Readiness requires status GREEN. + * + *

    In the {@code coordinator} role, the probe connects to the local coordinator server. A leader + * coordinator is Ready regardless of cluster color — its readiness must not depend on tablet + * health, and the tablet probes already gate on cluster recovery. A standby coordinator answers + * with its role and election state and is Ready only when the coordinator group currently has an + * elected leader, so "all coordinator pods Ready" certifies a functioning group. * *

    Inputs

    * @@ -53,9 +63,11 @@ * *