This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new ad51d97e8 [server] Add Cluster Health API implementation (#3400)
ad51d97e8 is described below
commit ad51d97e8d8f5e7e0ccf10a611fc68f21e916941
Author: yunhong <[email protected]>
AuthorDate: Wed Jun 10 23:21:49 2026 +0800
[server] Add Cluster Health API implementation (#3400)
---
.../java/org/apache/fluss/client/admin/Admin.java | 26 ++
.../apache/fluss/client/admin/ClusterHealth.java | 108 +++++++
.../fluss/client/admin/ClusterHealthStatus.java | 33 ++
.../org/apache/fluss/client/admin/FlussAdmin.java | 7 +
.../fluss/client/utils/ClientRpcMessageUtils.java | 25 ++
.../fluss/client/admin/FlussAdminITCase.java | 51 +++
.../java/org/apache/fluss/metrics/MetricNames.java | 1 +
.../src/main/resources/bin/readiness-check.sh | 271 ++++++++++++++++
.../flink/sink/testutils/TestAdminAdapter.java | 5 +
.../fluss/rpc/gateway/AdminReadOnlyGateway.java | 10 +
.../org/apache/fluss/rpc/protocol/ApiKeys.java | 3 +-
fluss-rpc/src/main/proto/FlussApi.proto | 9 +
.../fluss/rpc/TestingTabletGatewayService.java | 8 +
.../server/coordinator/CoordinatorContext.java | 78 ++++-
.../coordinator/CoordinatorEventProcessor.java | 10 +
.../coordinator/CoordinatorRequestBatch.java | 28 ++
.../server/coordinator/CoordinatorService.java | 57 ++++
.../coordinator/event/CoordinatorEventManager.java | 19 +-
.../apache/fluss/server/tablet/TabletServer.java | 4 +-
.../apache/fluss/server/tablet/TabletService.java | 35 ++-
.../server/tools/ClusterHealthReadinessCheck.java | 265 ++++++++++++++++
.../server/coordinator/ClusterHealthTest.java | 346 +++++++++++++++++++++
.../server/coordinator/CoordinatorContextTest.java | 127 ++++++++
.../coordinator/CoordinatorRequestBatchTest.java | 160 ++++++++++
.../server/coordinator/TestCoordinatorGateway.java | 8 +
.../server/tablet/TestTabletServerGateway.java | 8 +
fluss-test-coverage/pom.xml | 4 +-
helm/templates/sts-coordinator.yaml | 4 +-
helm/templates/sts-tablet.yaml | 32 +-
helm/values.yaml | 37 +++
website/docs/install-deploy/deploying-with-helm.md | 42 ++-
.../maintenance/observability/monitor-metrics.md | 9 +-
website/docs/maintenance/operations/upgrading.md | 38 +++
33 files changed, 1842 insertions(+), 26 deletions(-)
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 ea28a246b..5f0bcbde5 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
@@ -770,4 +770,30 @@ public interface Admin extends AutoCloseable {
* @since 0.9
*/
CompletableFuture<Void> deleteProducerOffsets(String producerId);
+
+ /**
+ * Get the health status of the cluster asynchronously.
+ *
+ * <p>The returned {@link ClusterHealth} contains replica statistics and
an overall {@link
+ * ClusterHealthStatus}:
+ *
+ * <ul>
+ * <li>{@link ClusterHealthStatus#GREEN} — all replicas are in-sync and
all leaders are
+ * active. The cluster is fully healthy.
+ * <li>{@link ClusterHealthStatus#YELLOW} — all leaders are active, but
some replicas have not
+ * yet rejoined the in-sync replica set (ISR).
+ * <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).
+ * </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.
+ *
+ * @return a {@link CompletableFuture} that completes with the cluster
health information.
+ * @since 1.0
+ */
+ CompletableFuture<ClusterHealth> getClusterHealth();
}
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
new file mode 100644
index 000000000..8f751bc16
--- /dev/null
+++
b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealth.java
@@ -0,0 +1,108 @@
+/*
+ * 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.client.admin;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+import java.util.Objects;
+
+/**
+ * Cluster health information returned by {@link Admin#getClusterHealth()}.
+ *
+ * @since 1.0
+ */
+@PublicEvolving
+public final class ClusterHealth {
+
+ private final int numReplicas;
+ private final int inSyncReplicas;
+ private final int numLeaderReplicas;
+ private final int activeLeaderReplicas;
+ private final ClusterHealthStatus status;
+
+ public ClusterHealth(
+ int numReplicas,
+ int inSyncReplicas,
+ int numLeaderReplicas,
+ int activeLeaderReplicas,
+ ClusterHealthStatus status) {
+ this.numReplicas = numReplicas;
+ this.inSyncReplicas = inSyncReplicas;
+ this.numLeaderReplicas = numLeaderReplicas;
+ this.activeLeaderReplicas = activeLeaderReplicas;
+ this.status = Objects.requireNonNull(status, "status");
+ }
+
+ public int getNumReplicas() {
+ return numReplicas;
+ }
+
+ public int getInSyncReplicas() {
+ return inSyncReplicas;
+ }
+
+ public int getNumLeaderReplicas() {
+ return numLeaderReplicas;
+ }
+
+ public int getActiveLeaderReplicas() {
+ return activeLeaderReplicas;
+ }
+
+ public ClusterHealthStatus getStatus() {
+ return status;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof ClusterHealth)) {
+ return false;
+ }
+ ClusterHealth that = (ClusterHealth) o;
+ return numReplicas == that.numReplicas
+ && inSyncReplicas == that.inSyncReplicas
+ && numLeaderReplicas == that.numLeaderReplicas
+ && activeLeaderReplicas == that.activeLeaderReplicas
+ && status == that.status;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ numReplicas, inSyncReplicas, numLeaderReplicas,
activeLeaderReplicas, status);
+ }
+
+ @Override
+ public String toString() {
+ return "ClusterHealth{"
+ + "numReplicas="
+ + numReplicas
+ + ", inSyncReplicas="
+ + inSyncReplicas
+ + ", numLeaderReplicas="
+ + numLeaderReplicas
+ + ", activeLeaderReplicas="
+ + activeLeaderReplicas
+ + ", status="
+ + status
+ + '}';
+ }
+}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealthStatus.java
b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealthStatus.java
new file mode 100644
index 000000000..531ebf0eb
--- /dev/null
+++
b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterHealthStatus.java
@@ -0,0 +1,33 @@
+/*
+ * 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.client.admin;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * Health status of the Fluss cluster.
+ *
+ * @since 1.0
+ */
+@PublicEvolving
+public enum ClusterHealthStatus {
+ GREEN,
+ YELLOW,
+ RED,
+ UNKNOWN
+}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
index 954247850..f724a15b8 100644
--- a/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
+++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java
@@ -68,6 +68,7 @@ import
org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DropAclsRequest;
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
import org.apache.fluss.rpc.messages.DropTableRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest;
import org.apache.fluss.rpc.messages.GetLakeSnapshotRequest;
@@ -896,6 +897,12 @@ public class FlussAdmin implements Admin {
}
}
+ @Override
+ public CompletableFuture<ClusterHealth> getClusterHealth() {
+ return gateway.getClusterHealth(new GetClusterHealthRequest())
+ .thenApply(ClientRpcMessageUtils::toClusterHealth);
+ }
+
@VisibleForTesting
public AdminGateway getAdminGateway() {
return gateway;
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 0bd67da17..3f6e2443a 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
@@ -17,6 +17,8 @@
package org.apache.fluss.client.utils;
+import org.apache.fluss.client.admin.ClusterHealth;
+import org.apache.fluss.client.admin.ClusterHealthStatus;
import org.apache.fluss.client.admin.OffsetSpec;
import org.apache.fluss.client.admin.ProducerOffsetsResult;
import org.apache.fluss.client.lookup.LookupBatch;
@@ -51,6 +53,7 @@ import org.apache.fluss.rpc.messages.AlterDatabaseRequest;
import org.apache.fluss.rpc.messages.AlterTableRequest;
import org.apache.fluss.rpc.messages.CreatePartitionRequest;
import org.apache.fluss.rpc.messages.DropPartitionRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse;
import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataResponse;
import org.apache.fluss.rpc.messages.GetLakeSnapshotResponse;
@@ -829,4 +832,26 @@ public class ClientRpcMessageUtils {
.collect(Collectors.toList());
return new
GetTableStatsRequest().setTableId(tableId).addAllBucketsReqs(pbBuckets);
}
+
+ public static ClusterHealth toClusterHealth(GetClusterHealthResponse resp)
{
+ return new ClusterHealth(
+ resp.getNumReplicas(),
+ resp.getInSyncReplicas(),
+ resp.getNumLeaderReplicas(),
+ resp.getActiveLeaderReplicas(),
+ toClusterHealthStatus(resp.getStatus()));
+ }
+
+ private static ClusterHealthStatus toClusterHealthStatus(int pbStatus) {
+ switch (pbStatus) {
+ case 0:
+ return ClusterHealthStatus.GREEN;
+ case 1:
+ return ClusterHealthStatus.YELLOW;
+ case 2:
+ return ClusterHealthStatus.RED;
+ default:
+ return ClusterHealthStatus.UNKNOWN;
+ }
+ }
}
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
index 65cca8195..68181844e 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java
@@ -2413,4 +2413,55 @@ class FlussAdminITCase extends ClientToServerITCaseBase {
.isInstanceOf(NetworkException.class)
.hasMessageContaining("connection timed out");
}
+
+ @Test
+ void testClusterHealthDuringRollingUpgrade() throws Exception {
+ TablePath tablePath = TablePath.of("test_db", "health_test_table");
+ TableDescriptor tableDescriptor =
+
TableDescriptor.builder().schema(DEFAULT_SCHEMA).distributedBy(3, "id").build();
+ long tableId = createTable(tablePath, tableDescriptor, true);
+ waitAllReplicasReady(tableId, 3);
+
+ // Phase 1: Cluster is healthy — status should be GREEN.
+ ClusterHealth health = admin.getClusterHealth().get();
+ assertThat(health.getStatus()).isEqualTo(ClusterHealthStatus.GREEN);
+
assertThat(health.getNumReplicas()).isEqualTo(health.getInSyncReplicas());
+
assertThat(health.getNumLeaderReplicas()).isEqualTo(health.getActiveLeaderReplicas());
+
+ // Phase 2: Stop one tablet server (simulate server crash during
rolling upgrade).
+ int stoppedServerId = 0;
+ FLUSS_CLUSTER_EXTENSION.stopTabletServer(stoppedServerId);
+ FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(2);
+
+ for (int bucket = 0; bucket < 3; bucket++) {
+ TableBucket tb = new TableBucket(tableId, bucket);
+ FLUSS_CLUSTER_EXTENSION.waitUntilReplicaShrinkFromIsr(tb,
stoppedServerId);
+ }
+
+ // Status should not be GREEN (YELLOW or RED depending on leader
placement).
+ ClusterHealth duringDown = admin.getClusterHealth().get();
+
assertThat(duringDown.getStatus()).isNotEqualTo(ClusterHealthStatus.GREEN);
+
assertThat(duringDown.getInSyncReplicas()).isLessThan(duringDown.getNumReplicas());
+
+ // Phase 3: Restart the server.
+ FLUSS_CLUSTER_EXTENSION.startTabletServer(stoppedServerId);
+ FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(3);
+
+ // Phase 4: Wait for recovery — status should return to GREEN.
+ for (int bucket = 0; bucket < 3; bucket++) {
+ TableBucket tb = new TableBucket(tableId, bucket);
+ FLUSS_CLUSTER_EXTENSION.waitUntilReplicaExpandToIsr(tb,
stoppedServerId);
+ }
+
+ waitUntil(
+ () -> admin.getClusterHealth().get().getStatus() ==
ClusterHealthStatus.GREEN,
+ Duration.ofMinutes(1),
+ "Cluster should return to GREEN after server restart");
+
+ ClusterHealth afterRecovery = admin.getClusterHealth().get();
+
assertThat(afterRecovery.getStatus()).isEqualTo(ClusterHealthStatus.GREEN);
+
assertThat(afterRecovery.getNumReplicas()).isEqualTo(afterRecovery.getInSyncReplicas());
+ assertThat(afterRecovery.getNumLeaderReplicas())
+ .isEqualTo(afterRecovery.getActiveLeaderReplicas());
+ }
}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
index fe34deba9..f56ca967c 100644
--- a/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
+++ b/fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java
@@ -44,6 +44,7 @@ public class MetricNames {
public static final String BUCKET_COUNT = "bucketCount";
public static final String PARTITION_COUNT = "partitionCount";
public static final String REPLICAS_TO_DELETE_COUNT =
"replicasToDeleteCount";
+ public static final String PENDING_LEADER_ACTIVATION_COUNT =
"pendingLeaderActivationCount";
// for coordinator event processor
public static final String EVENT_QUEUE_SIZE = "eventQueueSize";
diff --git a/fluss-dist/src/main/resources/bin/readiness-check.sh
b/fluss-dist/src/main/resources/bin/readiness-check.sh
new file mode 100644
index 000000000..7d2ac5d9a
--- /dev/null
+++ b/fluss-dist/src/main/resources/bin/readiness-check.sh
@@ -0,0 +1,271 @@
+#!/usr/bin/env bash
+
+#
+# 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.
+#
+
+#
==============================================================================
+# Fluss Readiness Check Script
+#
==============================================================================
+#
+# Two-step readiness probe for Kubernetes StatefulSet rolling upgrades:
+#
+# Step 1 — Local TCP port check: verify this TabletServer 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.
+#
+# Both steps must pass for the pod to be marked Ready.
+#
+# Exit codes (from ClusterHealthReadinessCheck.java):
+# 0 = Ready (status GREEN)
+# 1 = Not ready (status YELLOW, RED, UNKNOWN, or TabletServer unreachable)
+# 2 = API unsupported (older server) → immediate TCP fallback (latched)
+#
+# Environment variables (set by helm template or container spec):
+# FLUSS_HOME - Fluss installation directory
+# 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
+# binds the tablet's CLIENT endpoint to the pod IP rather than 0.0.0.0),
+# falling back to 127.0.0.1 otherwise. The probe always talks to the
local
+# sidecar tablet, which forwards getClusterHealth to the Coordinator over
+# the internal listener — so the probe never needs to know the
+# Coordinator's address.
+# READINESS_TCP_PORT - TabletServer client port (default: 9124).
+# READINESS_HEALTH_CHECK_AUTH - Optional client auth configuration as
+# semicolon-separated key:value pairs (e.g.
+# client.security.protocol:SASL;client.sasl.mechanism:PLAIN). Required
only
+# when the local client listener enforces SASL.
+# READINESS_HEALTH_CHECK_TIMEOUT_SECONDS - Maximum wall-clock seconds the
+# cluster health gate is allowed to keep this pod NotReady on first boot.
+# After this budget is exhausted the probe latches the TCP-only fast path
+# so a permanently-unrecoverable cluster cannot wedge a rolling upgrade.
+# Default: 1200 (20 minutes).
+#
==============================================================================
+
+set -o pipefail
+
+# ---- Configuration ----
+
+FLUSS_HOME="${FLUSS_HOME:-/opt/fluss}"
+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
+# Coordinator's address. Prefer ${POD_IP} when set (typical K8s case where
+# bind.listeners binds the CLIENT endpoint to the pod IP, not 0.0.0.0); fall
+# back to 127.0.0.1 for local/non-K8s invocations.
+TCP_HOST="${READINESS_TCP_HOST:-${POD_IP:-127.0.0.1}}"
+TCP_PORT="${READINESS_TCP_PORT:-9124}"
+# Wall-clock budget for the first-boot cluster health gate. Once exhausted, the
+# probe latches the TCP fast path so a permanently-unhealthy cluster cannot
+# wedge a rolling upgrade indefinitely.
+HEALTH_CHECK_TIMEOUT_SECONDS="${READINESS_HEALTH_CHECK_TIMEOUT_SECONDS:-1200}"
+
+# Marker files for tracking state across probe invocations
+MARKER_DIR="/tmp/fluss-readiness"
+FIRST_READY_MARKER="${MARKER_DIR}/first-ready"
+# Records the epoch of the first time this pod entered the cluster health gate.
+# Used to enforce HEALTH_CHECK_TIMEOUT_SECONDS across probe invocations (the
+# Java CLI is forked fresh each cycle, so the budget must live in a marker
+# file).
+FIRST_PROBE_EPOCH_FILE="${MARKER_DIR}/first-probe-epoch"
+# Latched-once marker so the "now in TCP-only fast path" notice is logged
+# exactly one time per pod lifetime (avoids flooding kubectl logs every 3s).
+FAST_PATH_LOGGED="${MARKER_DIR}/fast-path-logged"
+
+mkdir -p "${MARKER_DIR}"
+
+# ---- Helper Functions ----
+
+# Step 1: TCP port check (local liveness)
+check_tcp() {
+ local host="${TCP_HOST}"
+ local port="${1:-${TCP_PORT}}"
+ (echo > /dev/tcp/"${host}"/"${port}") >/dev/null 2>&1
+ return $?
+}
+
+# Step 2: Run the Java ClusterHealthReadinessCheck CLI tool (cluster health
check).
+# The tool ships inside fluss-server-${version}.jar; no extra jar
(fluss-client,
+# fluss-dist) is required.
+#
+# Host/port are passed explicitly via --host/--port (derived from TCP_HOST and
+# TCP_PORT above). Optional auth properties come from env var
+# READINESS_HEALTH_CHECK_AUTH (semicolon-separated key:value pairs).
+run_recovery_check() {
+ local timeout_ms="$1"
+
+ # Locate fluss-server jar (the only jar this probe needs).
+ local fluss_server_jar=""
+ while IFS= read -r -d '' jarfile; do
+ if [[ "$jarfile" =~ .*/fluss-server[^/]*.jar$ ]]; then
+ fluss_server_jar="$jarfile"
+ break
+ fi
+ done < <(find "${FLUSS_HOME}/lib" ! -type d -name '*.jar' -print0 | sort
-z)
+
+ if [[ -z "$fluss_server_jar" ]]; then
+ echo "[readiness-check] ERROR: fluss-server jar not found in
${FLUSS_HOME}/lib"
+ return 3
+ fi
+
+ # log4j-api lives next to fluss-server jar in the dist; include the whole
lib/
+ # on the classpath so SLF4J/log4j wiring resolves correctly.
+ local classpath="${FLUSS_HOME}/lib/*"
+
+ # Find Java
+ local java_cmd="java"
+ if [[ -n "${JAVA_HOME}" && -x "${JAVA_HOME}/bin/java" ]]; then
+ java_cmd="${JAVA_HOME}/bin/java"
+ fi
+
+ # Run the check — host/port are explicit CLI flags; optional auth is read
+ # from READINESS_HEALTH_CHECK_AUTH env (already in the process
environment).
+ local output
+ output=$("${java_cmd}" \
+ -XX:+IgnoreUnrecognizedVMOptions \
+ -Xmx64m \
+ -classpath "${classpath}" \
+ org.apache.fluss.server.tools.ClusterHealthReadinessCheck \
+ --host "${TCP_HOST}" \
+ --port "${TCP_PORT}" \
+ --timeoutMs "${timeout_ms}" 2>&1)
+ local exit_code=$?
+
+ # Log output to container's main process stderr (visible in kubectl logs)
+ if [[ -n "$output" ]]; then
+ echo "$output" >&2
+ if [[ -w /proc/1/fd/2 ]]; then
+ echo "[readiness-probe] $output" > /proc/1/fd/2
+ fi
+ fi
+
+ return $exit_code
+}
+
+# ---- Main Logic ----
+
+# Helper: log to container's main process (visible in kubectl logs)
+log_to_main() {
+ echo "$1" >&2
+ if [[ -w /proc/1/fd/2 ]]; then
+ echo "$1" > /proc/1/fd/2
+ fi
+}
+
+# ---- Step 1: Local TCP port check ----
+# If the local TS process hasn't bound its port yet, fail immediately.
+# No need to query the Coordinator for cluster state.
+if ! check_tcp; then
+ exit 1
+fi
+
+# ---- Steady-state fast path ----
+#
+# K8s readinessProbe runs every periodSeconds FOREVER, not only during a
+# rolling upgrade. The cluster health gate (Step 2 below) is meant to gate
+# the FIRST readiness of a freshly created pod so that recovery finishes
+# before traffic flows. Running it on every probe cycle is unsafe:
+#
+# 1. Coordinator pod briefly goes down (e.g. its own rolling restart).
+# 2. ALL tablet-server probes call run_recovery_check at the same period,
+# all fail to reach Coordinator → all flip to NotReady simultaneously.
+# 3. statefulset then rolls all tablet-servers together → cascade outage:
+# coordinator-server-0 0/2 Init:0/1
+# tablet-server-0/1/2 0/2 Init:0/1
+#
+# So once the cluster has been GREEN at least once in this pod's lifetime
+# (FIRST_READY_MARKER present), we DROP to TCP-only. The marker lives in
+# /tmp/fluss-readiness, which is wiped whenever the pod is recreated by an
+# upgrade — so each freshly created pod will run the recovery gate exactly
+# ONCE before flipping to the fast path.
+if [[ -f "${FIRST_READY_MARKER}" ]]; then
+ # Log the switch-over notice exactly once per pod lifetime so operators
+ # can confirm via `kubectl logs` that this pod has graduated from the
+ # cluster health gate to the cheap TCP-only port check.
+ if [[ ! -f "${FAST_PATH_LOGGED}" ]]; then
+ log_to_main "[readiness-check] Switched to TCP-only port-readiness
check; cluster health gate will not run again until this pod is recreated"
+ touch "${FAST_PATH_LOGGED}"
+ fi
+ exit 0
+fi
+
+# ---- Step 2: Cluster health check (first boot of this pod only) ----
+#
+# We reach here only if this pod has never been ready in its lifetime. Block
+# traffic until the cluster reports GREEN. Once we latch the marker,
+# subsequent probes take the fast path above.
+
+# Record the epoch of the first probe attempt so we can enforce the recovery
+# budget across subsequent (forked-fresh) Java CLI invocations.
+if [[ ! -f "${FIRST_PROBE_EPOCH_FILE}" ]]; then
+ date +%s > "${FIRST_PROBE_EPOCH_FILE}"
+fi
+first_probe_epoch=$(cat "${FIRST_PROBE_EPOCH_FILE}" 2>/dev/null || echo 0)
+now_epoch=$(date +%s)
+elapsed_seconds=$(( now_epoch - first_probe_epoch ))
+
+# Bail out of the gate if the cluster has stayed unhealthy for too long. We
+# explicitly trade off "do not serve traffic from a half-recovered server" for
+# "never wedge a rolling upgrade" — operators can lengthen the budget via
+# READINESS_HEALTH_CHECK_TIMEOUT_SECONDS if they want stricter behaviour.
+if (( elapsed_seconds >= HEALTH_CHECK_TIMEOUT_SECONDS )); then
+ log_to_main "[readiness-check] Cluster health gate budget exhausted
(${elapsed_seconds}s >= ${HEALTH_CHECK_TIMEOUT_SECONDS}s); latching TCP fast
path to unblock rolling upgrade"
+ touch "${FIRST_READY_MARKER}"
+ exit 0
+fi
+
+run_recovery_check "${TIMEOUT_MS}"
+local_exit=$?
+
+case $local_exit in
+ 0)
+ # Recovery completed — latch fast path for the rest of this pod's life.
+ log_to_main "[readiness-check] Cluster health GREEN; latching fast
path — next probe will switch to TCP-only port check"
+ touch "${FIRST_READY_MARKER}"
+ exit 0
+ ;;
+ 1)
+ # Not recovered yet (or Coordinator temporarily unreachable) — stay
+ # unready. Next probe cycle will retry. This is the upgrade gate.
+ log_to_main "[readiness-check] First boot: BLOCKED (exit 1) — waiting
for recovery"
+ exit 1
+ ;;
+ 2)
+ # API unsupported (older Coordinator that predates the Cluster Health
+ # API). Server is up but definitively cannot answer; there is nothing
+ # to wait for, so latch the fast path immediately and rely on TCP-only
+ # readiness from now on. Waiting here would waste a grace period ×
+ # pod_count of dead time on the first upgrade from such a version,
+ # with zero protection gained (the API genuinely cannot be available
+ # while the entire cluster is on the old Coordinator).
+ log_to_main "[readiness-check] Cluster Health API unsupported by
Coordinator, TCP fallback (latched)"
+ touch "${FIRST_READY_MARKER}"
+ exit 0
+ ;;
+ *)
+ # Configuration error — fall back to TCP and latch fast path so a
+ # broken probe never wedges the rolling upgrade forever.
+ log_to_main "[readiness-check] Config error (exit $local_exit), TCP
fallback (latched)"
+ touch "${FIRST_READY_MARKER}"
+ exit 0
+ ;;
+esac
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java
index 708226545..b8454c713 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java
@@ -317,4 +317,9 @@ public class TestAdminAdapter implements Admin {
public CompletableFuture<LakeSnapshot> getReadableLakeSnapshot(TablePath
tablePath) {
throw new UnsupportedOperationException("Not implemented in
TestAdminAdapter");
}
+
+ @Override
+ public CompletableFuture<org.apache.fluss.client.admin.ClusterHealth>
getClusterHealth() {
+ throw new UnsupportedOperationException("Not implemented in
TestAdminAdapter");
+ }
}
diff --git
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java
index 272d1b4a1..574e1a510 100644
---
a/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java
+++
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java
@@ -22,6 +22,8 @@ import org.apache.fluss.rpc.messages.DatabaseExistsRequest;
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
import org.apache.fluss.rpc.messages.GetDatabaseInfoResponse;
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenRequest;
@@ -192,4 +194,12 @@ public interface AdminReadOnlyGateway extends RpcGateway {
@RPC(api = ApiKeys.DESCRIBE_CLUSTER_CONFIGS)
CompletableFuture<DescribeClusterConfigsResponse> describeClusterConfigs(
DescribeClusterConfigsRequest request);
+
+ /**
+ * Get the overall cluster health snapshot.
+ *
+ * @return the cluster health response.
+ */
+ @RPC(api = ApiKeys.GET_CLUSTER_HEALTH)
+ CompletableFuture<GetClusterHealthResponse>
getClusterHealth(GetClusterHealthRequest request);
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
index 92a4680ee..dd164d7cc 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java
@@ -103,7 +103,8 @@ public enum ApiKeys {
DROP_KV_SNAPSHOT_LEASE(1058, 0, 0, PUBLIC),
GET_TABLE_STATS(1059, 0, 0, PUBLIC),
ALTER_DATABASE(1060, 0, 0, PUBLIC),
- SCAN_KV(1061, 0, 0, PUBLIC);
+ SCAN_KV(1061, 0, 0, PUBLIC),
+ GET_CLUSTER_HEALTH(1062, 0, 0, PUBLIC);
private static final Map<Integer, ApiKeys> ID_TO_TYPE =
Arrays.stream(ApiKeys.values())
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto
b/fluss-rpc/src/main/proto/FlussApi.proto
index e8381215f..1e66bc343 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -767,6 +767,15 @@ message DeleteProducerOffsetsRequest {
message DeleteProducerOffsetsResponse {
}
+message GetClusterHealthRequest { }
+
+message GetClusterHealthResponse {
+ required int32 num_replicas = 1;
+ required int32 in_sync_replicas = 2;
+ 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
+}
// --------------- Inner classes ----------------
diff --git
a/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java
b/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java
index 351768093..f465bb4a6 100644
---
a/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java
+++
b/fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java
@@ -25,6 +25,8 @@ import
org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import org.apache.fluss.rpc.messages.FetchLogResponse;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
import org.apache.fluss.rpc.messages.GetDatabaseInfoResponse;
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenRequest;
@@ -264,4 +266,10 @@ public class TestingTabletGatewayService extends
TestingGatewayService
public CompletableFuture<ScanKvResponse> scanKv(ScanKvRequest request) {
return null;
}
+
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ return null;
+ }
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
index 92dfc5ee5..dc0fa58a3 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java
@@ -106,6 +106,13 @@ public class CoordinatorContext {
*/
private final Map<Integer, Set<TableBucket>> replicasOnOffline = new
HashMap<>();
+ /**
+ * Tracks buckets where a leader change has been dispatched (via
NotifyLeaderAndIsr) but not yet
+ * confirmed by the target server. A bucket enters this set when we send
the notification and
+ * leaves it when the target server successfully responds confirming it is
the leader.
+ */
+ private final Set<TableBucket> pendingLeaderActivationBuckets = new
HashSet<>();
+
/** A mapping from tabletServers to server tag. */
private final Map<Integer, ServerTag> serverTags = new HashMap<>();
@@ -218,6 +225,48 @@ public class CoordinatorContext {
replicasOnOffline.remove(serverId);
}
+ // ---- Pending leader activation tracking (for Cluster Health API) ----
+
+ public void addPendingLeaderActivation(TableBucket bucket) {
+ pendingLeaderActivationBuckets.add(bucket);
+ }
+
+ public void addPendingLeaderActivations(Collection<TableBucket> buckets) {
+ pendingLeaderActivationBuckets.addAll(buckets);
+ }
+
+ public void clearPendingLeaderActivation(TableBucket bucket) {
+ pendingLeaderActivationBuckets.remove(bucket);
+ }
+
+ /**
+ * Returns whether the given bucket has an active leader. A leader is
considered active when:
+ *
+ * <ul>
+ * <li>A LeaderAndIsr record exists for the bucket
+ * <li>The leader is not {@link LeaderAndIsr#NO_LEADER}
+ * <li>The leader's tablet server is alive
+ * <li>The bucket is not pending leader activation confirmation
+ * </ul>
+ */
+ public boolean isLeaderActive(TableBucket bucket) {
+ return getBucketLeaderAndIsr(bucket)
+ .map(
+ lai ->
+ lai.leader() != LeaderAndIsr.NO_LEADER
+ &&
liveTabletServers.containsKey(lai.leader())
+ &&
!pendingLeaderActivationBuckets.contains(bucket))
+ .orElse(false);
+ }
+
+ public Set<TableBucket> getPendingLeaderActivationBuckets() {
+ return Collections.unmodifiableSet(pendingLeaderActivationBuckets);
+ }
+
+ public void removeFromPendingLeaderActivations(Set<TableBucket> buckets) {
+ pendingLeaderActivationBuckets.removeAll(buckets);
+ }
+
public Map<Long, TablePath> allTables() {
return tablePathById;
}
@@ -656,10 +705,16 @@ public class CoordinatorContext {
tablesToBeDeleted.remove(tableId);
Map<Integer, List<Integer>> assignment =
tableAssignments.remove(tableId);
if (assignment != null) {
- // remove leadership info for each bucket from the context
+ Set<TableBucket> removedBuckets = new HashSet<>();
assignment
.keySet()
- .forEach(bucket -> bucketLeaderAndIsr.remove(new
TableBucket(tableId, bucket)));
+ .forEach(
+ bucket -> {
+ TableBucket tb = new TableBucket(tableId,
bucket);
+ bucketLeaderAndIsr.remove(tb);
+ removedBuckets.add(tb);
+ });
+ removeFromPendingLeaderActivations(removedBuckets);
}
TablePath tablePath = tablePathById.remove(tableId);
@@ -673,16 +728,20 @@ public class CoordinatorContext {
partitionsToBeDeleted.remove(tablePartition);
Map<Integer, List<Integer>> assignment =
partitionAssignments.remove(tablePartition);
if (assignment != null) {
- // remove leadership info for each bucket from the context
+ Set<TableBucket> removedBuckets = new HashSet<>();
assignment
.keySet()
.forEach(
- bucket ->
- bucketLeaderAndIsr.remove(
- new TableBucket(
-
tablePartition.getTableId(),
-
tablePartition.getPartitionId(),
- bucket)));
+ bucket -> {
+ TableBucket tb =
+ new TableBucket(
+ tablePartition.getTableId(),
+
tablePartition.getPartitionId(),
+ bucket);
+ bucketLeaderAndIsr.remove(tb);
+ removedBuckets.add(tb);
+ });
+ removeFromPendingLeaderActivations(removedBuckets);
}
PhysicalTablePath physicalTablePath =
@@ -717,6 +776,7 @@ public class CoordinatorContext {
partitionAssignments.clear();
bucketLeaderAndIsr.clear();
replicasOnOffline.clear();
+ pendingLeaderActivationBuckets.clear();
bucketStates.clear();
replicaStates.clear();
tablePathById.clear();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 62d63ae1f..13eaba282 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -1003,6 +1003,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// get the server that receives the response
int serverId =
notifyLeaderAndIsrResponseReceivedEvent.getResponseServerId();
Set<TableBucketReplica> offlineReplicas = new HashSet<>();
+ List<TableBucket> succeededBuckets = new ArrayList<>();
// get all the results for each bucket
List<NotifyLeaderAndIsrResultForBucket>
notifyLeaderAndIsrResultForBuckets =
notifyLeaderAndIsrResponseReceivedEvent.getNotifyLeaderAndIsrResultForBuckets();
@@ -1013,6 +1014,14 @@ public class CoordinatorEventProcessor implements
EventProcessor {
offlineReplicas.add(
new TableBucketReplica(
notifyLeaderAndIsrResultForBucket.getTableBucket(), serverId));
+ } else {
+
succeededBuckets.add(notifyLeaderAndIsrResultForBucket.getTableBucket());
+ }
+ }
+ for (TableBucket tb : succeededBuckets) {
+ Optional<LeaderAndIsr> laiOpt =
coordinatorContext.getBucketLeaderAndIsr(tb);
+ if (laiOpt.isPresent() && laiOpt.get().leader() == serverId) {
+ coordinatorContext.clearPendingLeaderActivation(tb);
}
}
if (!offlineReplicas.isEmpty()) {
@@ -1169,6 +1178,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// process dead tablet server
LOG.info("Tablet server failure callback for {}.", tabletServerId);
coordinatorContext.removeOfflineBucketInServer(tabletServerId);
+
coordinatorContext.removeLiveTabletServer(tabletServerId);
coordinatorContext.shuttingDownTabletServers().remove(tabletServerId);
coordinatorChannelManager.removeTabletServer(tabletServerId);
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
index 344105367..3ec69feeb 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java
@@ -35,6 +35,7 @@ import
org.apache.fluss.rpc.messages.PbStopReplicaRespForBucket;
import org.apache.fluss.rpc.messages.StopReplicaRequest;
import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
import org.apache.fluss.rpc.protocol.ApiError;
+import org.apache.fluss.server.coordinator.event.AccessContextEvent;
import
org.apache.fluss.server.coordinator.event.DeleteReplicaResponseReceivedEvent;
import org.apache.fluss.server.coordinator.event.EventManager;
import
org.apache.fluss.server.coordinator.event.NotifyLeaderAndIsrResponseReceivedEvent;
@@ -413,6 +414,18 @@ public class CoordinatorRequestBatch {
makeNotifyLeaderAndIsrRequest(
coordinatorEpoch,
notifyRequestEntry.getValue().values());
+ // Track exactly which buckets THIS request marked as pending
leader activation. Only
+ // those entries (where leader == serverId) need to be cleared if
the request fails
+ Set<TableBucket> addedToPendingLeaderActivation = new HashSet<>();
+ for (Map.Entry<TableBucket, PbNotifyLeaderAndIsrReqForBucket>
entry :
+ notifyRequestEntry.getValue().entrySet()) {
+ int leader = entry.getValue().getLeader();
+ if (leader == serverId) {
+
coordinatorContext.addPendingLeaderActivation(entry.getKey());
+ addedToPendingLeaderActivation.add(entry.getKey());
+ }
+ }
+
coordinatorChannelManager.sendBucketLeaderAndIsrRequest(
serverId,
notifyLeaderAndIsrRequest,
@@ -428,6 +441,21 @@ public class CoordinatorRequestBatch {
// coordinator will remove the sender for the
tablet server and mark all
// replica in the tablet server as offline. so, in
here, if encounter
// any error, we just ignore it.
+
+ // Clear pending state so the health API does not
report stale
+ // RED. The coordinator will detect actual server
death via
+ // heartbeat timeout and trigger re-election
separately.
+ if (!addedToPendingLeaderActivation.isEmpty()) {
+ eventManager.put(
+ new AccessContextEvent<Void>(
+ ctx -> {
+ for (TableBucket tb :
+
addedToPendingLeaderActivation) {
+
ctx.clearPendingLeaderActivation(tb);
+ }
+ return null;
+ }));
+ }
return;
}
// put the response receive event into the event
manager
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 5a37aa9a6..8c122cb78 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
@@ -98,6 +98,8 @@ import org.apache.fluss.rpc.messages.DropPartitionRequest;
import org.apache.fluss.rpc.messages.DropPartitionResponse;
import org.apache.fluss.rpc.messages.DropTableRequest;
import org.apache.fluss.rpc.messages.DropTableResponse;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetProducerOffsetsRequest;
import org.apache.fluss.rpc.messages.GetProducerOffsetsResponse;
import org.apache.fluss.rpc.messages.LakeTieringHeartbeatRequest;
@@ -166,6 +168,7 @@ import
org.apache.fluss.server.metadata.CoordinatorMetadataProvider;
import org.apache.fluss.server.utils.ServerRpcMessageUtils;
import org.apache.fluss.server.zk.ZooKeeperClient;
import org.apache.fluss.server.zk.data.BucketAssignment;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.server.zk.data.PartitionAssignment;
import org.apache.fluss.server.zk.data.TableAssignment;
import org.apache.fluss.server.zk.data.TableRegistration;
@@ -1234,6 +1237,60 @@ public final class CoordinatorService extends
RpcServiceBase implements Coordina
return response;
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ if (authorizer != null) {
+ authorizer.authorize(currentSession(), OperationType.DESCRIBE,
Resource.cluster());
+ }
+
+ AccessContextEvent<GetClusterHealthResponse> event =
+ new
AccessContextEvent<>(CoordinatorService::computeClusterHealth);
+ eventManagerSupplier.get().put(event);
+ return event.getResultFuture();
+ }
+
+ @VisibleForTesting
+ static GetClusterHealthResponse computeClusterHealth(CoordinatorContext
ctx) {
+ GetClusterHealthResponse response = new GetClusterHealthResponse();
+
+ int numReplicas = 0;
+ int inSyncReplicas = 0;
+ int numLeaderReplicas = 0;
+ int activeLeaderReplicas = 0;
+
+ for (TableBucket tb : ctx.getAllBuckets()) {
+ List<Integer> assignment = ctx.getAssignment(tb);
+ numReplicas += assignment.size();
+ numLeaderReplicas++;
+
+ Optional<LeaderAndIsr> laiOpt = ctx.getBucketLeaderAndIsr(tb);
+ if (laiOpt.isPresent()) {
+ inSyncReplicas += laiOpt.get().isr().size();
+ }
+ if (ctx.isLeaderActive(tb)) {
+ activeLeaderReplicas++;
+ }
+ }
+
+ // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
+ int status;
+ if (activeLeaderReplicas < numLeaderReplicas) {
+ status = 2; // RED
+ } else if (inSyncReplicas < numReplicas) {
+ status = 1; // YELLOW
+ } else {
+ status = 0; // GREEN
+ }
+
+ response.setNumReplicas(numReplicas);
+ response.setInSyncReplicas(inSyncReplicas);
+ response.setNumLeaderReplicas(numLeaderReplicas);
+ response.setActiveLeaderReplicas(activeLeaderReplicas);
+ response.setStatus(status);
+ return response;
+ }
+
@VisibleForTesting
public DataLakeFormat getDataLakeFormat() {
return
lakeCatalogDynamicLoader.getLakeCatalogContainer().getDataLakeFormat();
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
index 6f13df501..33df6f484 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/CoordinatorEventManager.java
@@ -72,6 +72,12 @@ public final class CoordinatorEventManager implements
EventManager {
private volatile int partitionCount;
private volatile int replicasToDeleteCount;
+ /**
+ * Number of buckets currently waiting for the leader-activation ack from
the target tablet
+ * server.
+ */
+ private volatile int pendingLeaderActivationCount;
+
private static final int WINDOW_SIZE = 100;
private static final long METRICS_UPDATE_INTERVAL_MS = 5000; // 5 seconds
@@ -101,6 +107,8 @@ public final class CoordinatorEventManager implements
EventManager {
coordinatorMetricGroup.gauge(MetricNames.PARTITION_COUNT, () ->
partitionCount);
coordinatorMetricGroup.gauge(
MetricNames.REPLICAS_TO_DELETE_COUNT, () ->
replicasToDeleteCount);
+ coordinatorMetricGroup.gauge(
+ MetricNames.PENDING_LEADER_ACTIVATION_COUNT, () ->
pendingLeaderActivationCount);
}
/** Not thread safety! this method can only be executed in the
CoordinatorEventThread. */
@@ -116,6 +124,8 @@ public final class CoordinatorEventManager implements
EventManager {
int bucketCount =
context.bucketLeaderAndIsr().size();
int partitionCount =
context.getTotalPartitionCount();
int offlineBucketCount =
context.getOfflineBucketCount();
+ int pendingLeaderActivationCount =
+
context.getPendingLeaderActivationBuckets().size();
int replicasToDeletes = 0;
// for replica in partitions to be deleted
@@ -150,7 +160,8 @@ public final class CoordinatorEventManager implements
EventManager {
bucketCount,
partitionCount,
offlineBucketCount,
- replicasToDeletes);
+ replicasToDeletes,
+ pendingLeaderActivationCount);
});
eventProcessor.process(accessContextEvent);
@@ -166,6 +177,7 @@ public final class CoordinatorEventManager implements
EventManager {
this.partitionCount = metricsData.partitionCount;
this.offlineBucketCount = metricsData.offlineBucketCount;
this.replicasToDeleteCount = metricsData.replicasToDeleteCount;
+ this.pendingLeaderActivationCount =
metricsData.pendingLeaderActivationCount;
} catch (Exception e) {
LOG.warn("Failed to update metrics via AccessContextEvent", e);
}
@@ -298,6 +310,7 @@ public final class CoordinatorEventManager implements
EventManager {
private final int partitionCount;
private final int offlineBucketCount;
private final int replicasToDeleteCount;
+ private final int pendingLeaderActivationCount;
public MetricsData(
int coordinatorServerCount,
@@ -307,7 +320,8 @@ public final class CoordinatorEventManager implements
EventManager {
int bucketCount,
int partitionCount,
int offlineBucketCount,
- int replicasToDeleteCount) {
+ int replicasToDeleteCount,
+ int pendingLeaderActivationCount) {
this.coordinatorServerCount = coordinatorServerCount;
this.tabletServerCount = tabletServerCount;
this.tableCount = tableCount;
@@ -316,6 +330,7 @@ public final class CoordinatorEventManager implements
EventManager {
this.partitionCount = partitionCount;
this.offlineBucketCount = offlineBucketCount;
this.replicasToDeleteCount = replicasToDeleteCount;
+ this.pendingLeaderActivationCount = pendingLeaderActivationCount;
}
}
}
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
index 74afa8337..c7c5fdd78 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
@@ -315,7 +315,9 @@ public class TabletServer extends ServerBase {
authorizer,
dynamicConfigManager,
ioExecutor,
- scannerManager);
+ scannerManager,
+ coordinatorGateway,
+ interListenerName);
RequestsMetrics requestsMetrics =
RequestsMetrics.createTabletServerRequestMetrics(tabletServerMetricGroup);
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
index bdeff1818..892bcbc34 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletService.java
@@ -22,6 +22,7 @@ import org.apache.fluss.exception.AuthorizationException;
import org.apache.fluss.exception.InvalidScanRequestException;
import org.apache.fluss.exception.NotLeaderOrFollowerException;
import org.apache.fluss.exception.ScannerExpiredException;
+import org.apache.fluss.exception.StaleMetadataException;
import org.apache.fluss.exception.UnknownScannerIdException;
import org.apache.fluss.exception.UnknownTableOrBucketException;
import org.apache.fluss.fs.FileSystem;
@@ -34,9 +35,12 @@ import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.entity.LookupResultForBucket;
import org.apache.fluss.rpc.entity.PrefixLookupResultForBucket;
import org.apache.fluss.rpc.entity.ResultForBucket;
+import org.apache.fluss.rpc.gateway.CoordinatorGateway;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import org.apache.fluss.rpc.messages.FetchLogResponse;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetTableStatsRequest;
import org.apache.fluss.rpc.messages.GetTableStatsResponse;
import org.apache.fluss.rpc.messages.InitWriterRequest;
@@ -148,6 +152,8 @@ public final class TabletService extends RpcServiceBase
implements TabletServerG
private final TabletServerMetadataCache metadataCache;
private final TabletServerMetadataProvider metadataFunctionProvider;
private final ScannerManager scannerManager;
+ private final CoordinatorGateway coordinatorGateway;
+ private final String interListenerName;
public TabletService(
int serverId,
@@ -159,7 +165,9 @@ public final class TabletService extends RpcServiceBase
implements TabletServerG
@Nullable Authorizer authorizer,
DynamicConfigManager dynamicConfigManager,
ExecutorService ioExecutor,
- ScannerManager scannerManager) {
+ ScannerManager scannerManager,
+ CoordinatorGateway coordinatorGateway,
+ String interListenerName) {
super(
remoteFileSystem,
ServerType.TABLET_SERVER,
@@ -174,6 +182,8 @@ public final class TabletService extends RpcServiceBase
implements TabletServerG
this.metadataFunctionProvider =
new TabletServerMetadataProvider(zkClient, metadataManager,
metadataCache);
this.scannerManager = scannerManager;
+ this.coordinatorGateway = coordinatorGateway;
+ this.interListenerName = interListenerName;
}
@Override
@@ -446,6 +456,29 @@ public final class TabletService extends RpcServiceBase
implements TabletServerG
return response;
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ // Tablet servers don't own the cluster-wide health view; we forward
the call to the
+ // coordinator over the internal listener.
+ if (authorizer != null) {
+ authorizer.authorize(currentSession(), OperationType.DESCRIBE,
Resource.cluster());
+ }
+
+ if (metadataCache.getCoordinatorServer(interListenerName) == null) {
+ // Fail fast during the startup window before the tablet has
received its first
+ // UpdateMetadataRequest from the coordinator. The supplier inside
coordinatorGateway
+ // would otherwise block-and-retry, which is not what readiness
probes want.
+ CompletableFuture<GetClusterHealthResponse> failed = new
CompletableFuture<>();
+ failed.completeExceptionally(
+ new StaleMetadataException(
+ "Tablet server has not yet received coordinator
metadata; cluster"
+ + " health is unavailable."));
+ return failed;
+ }
+ return coordinatorGateway.getClusterHealth(request);
+ }
+
@Override
public CompletableFuture<ScanKvResponse> scanKv(ScanKvRequest request) {
ScanKvResponse response = new ScanKvResponse();
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
new file mode 100644
index 000000000..320afbc60
--- /dev/null
+++
b/fluss-server/src/main/java/org/apache/fluss/server/tools/ClusterHealthReadinessCheck.java
@@ -0,0 +1,265 @@
+/*
+ * 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.tools;
+
+import org.apache.fluss.cluster.ServerNode;
+import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.config.Configuration;
+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.TabletServerGateway;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
+import org.apache.fluss.rpc.metrics.ClientMetricGroup;
+import org.apache.fluss.utils.ExceptionUtils;
+
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Lightweight readiness-check CLI for Fluss tablet-server pods.
+ *
+ * <p>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.
+ *
+ * <h3>Inputs</h3>
+ *
+ * <p>For every input below, a CLI argument (when supplied) takes precedence
over the corresponding
+ * environment variable. The env vars exist so the Kubernetes manifest can
stay declarative; the CLI
+ * flags exist so an operator can override them ad-hoc when running the probe
by hand.
+ *
+ * <ul>
+ * <li>{@code --timeoutMs <ms>}: optional, defaults to {@value
#DEFAULT_TIMEOUT_MS}.
+ * <li>{@code --host <host>} / env {@value #ENV_TCP_HOST}: tablet server
host. Defaults to {@code
+ * 127.0.0.1}.
+ * <li>{@code --port <port>} / env {@value #ENV_TCP_PORT}: tablet server
port. Defaults to {@value
+ * #DEFAULT_TCP_PORT}.
+ * <li>{@code --healthCheckAuth <props>} / env {@value
#ENV_HEALTH_CHECK_AUTH}: optional client
+ * auth configuration as semicolon-separated {@code key:value} pairs
(e.g. {@code
+ * client.security.protocol:SASL;client.sasl.mechanism:PLAIN;
+ *
client.security.sasl.username:admin;client.security.sasl.password:admin-pass}).
When
+ * absent, the probe uses unauthenticated PLAINTEXT.
+ * </ul>
+ *
+ * <h3>Exit codes</h3>
+ *
+ * <ul>
+ * <li>{@value #EXIT_READY} — cluster status is GREEN (Ready).
+ * <li>{@value #EXIT_NOT_READY} — cluster status is YELLOW/RED/UNKNOWN, the
tablet server is
+ * unreachable, or the RPC timed out.
+ * <li>{@value #EXIT_API_UNSUPPORTED} — tablet server is reachable but does
not implement the
+ * cluster-health API ({@link UnsupportedVersionException}).
+ * <li>{@value #EXIT_ERROR} — invalid arguments or environment.
+ * </ul>
+ */
+public final class ClusterHealthReadinessCheck {
+
+ public static final int EXIT_READY = 0;
+ public static final int EXIT_NOT_READY = 1;
+ public static final int EXIT_API_UNSUPPORTED = 2;
+ public static final int EXIT_ERROR = 3;
+
+ private static final long DEFAULT_TIMEOUT_MS = 5000;
+ private static final String DEFAULT_TCP_HOST = "127.0.0.1";
+ private static final int DEFAULT_TCP_PORT = 9124;
+
+ /** Environment variable carrying the tablet server host. */
+ static final String ENV_TCP_HOST = "READINESS_TCP_HOST";
+
+ /** Environment variable carrying the tablet server port. */
+ static final String ENV_TCP_PORT = "READINESS_TCP_PORT";
+
+ /**
+ * Environment variable carrying client auth configuration as
semicolon-separated {@code
+ * key:value} pairs. Values are forwarded verbatim to {@link
Configuration} before {@link
+ * RpcClient#create} is called.
+ */
+ static final String ENV_HEALTH_CHECK_AUTH = "READINESS_HEALTH_CHECK_AUTH";
+
+ /**
+ * GREEN status code as produced by {@code
CoordinatorService#computeClusterHealth} (mirrors
+ * {@code PbClusterHealthStatus.GREEN}).
+ */
+ private static final int STATUS_GREEN = 0;
+
+ private ClusterHealthReadinessCheck() {}
+
+ public static void main(String[] args) {
+ System.exit(run(args));
+ }
+
+ static int run(String[] args) {
+ long timeoutMs = DEFAULT_TIMEOUT_MS;
+ String hostArg = null;
+ String portArg = null;
+ String authArg = null;
+ for (int i = 0; i < args.length; i++) {
+ if ("--timeoutMs".equals(args[i]) && i + 1 < args.length) {
+ String raw = args[++i];
+ try {
+ timeoutMs = Long.parseLong(raw);
+ } catch (NumberFormatException e) {
+ System.err.println(
+ "[readiness-check] ERROR: invalid --timeoutMs
value: " + raw);
+ return EXIT_ERROR;
+ }
+ } else if ("--host".equals(args[i]) && i + 1 < args.length) {
+ hostArg = args[++i];
+ } else if ("--port".equals(args[i]) && i + 1 < args.length) {
+ portArg = args[++i];
+ } else if ("--healthCheckAuth".equals(args[i]) && i + 1 <
args.length) {
+ authArg = args[++i];
+ }
+ }
+
+ // CLI flag takes precedence over env so operators can override ad-hoc.
+ String host = firstNonBlank(hostArg, System.getenv(ENV_TCP_HOST),
DEFAULT_TCP_HOST);
+ String portStr = firstNonBlank(portArg, System.getenv(ENV_TCP_PORT),
null);
+ int port = DEFAULT_TCP_PORT;
+ if (portStr != null) {
+ try {
+ port = Integer.parseInt(portStr.trim());
+ } catch (NumberFormatException e) {
+ System.err.println("[readiness-check] ERROR: invalid port \""
+ portStr + "\"");
+ return EXIT_ERROR;
+ }
+ }
+ // The server id is irrelevant for a one-shot probe; pick a sentinel
value.
+ ServerNode tabletServer = new ServerNode(0, host, port,
ServerType.TABLET_SERVER);
+ String address = host + ":" + port;
+
+ String authPropsString = authArg != null ? authArg :
System.getenv(ENV_HEALTH_CHECK_AUTH);
+ Configuration conf;
+ try {
+ conf = buildConfiguration(authPropsString);
+ } catch (IllegalArgumentException e) {
+ System.err.println(
+ "[readiness-check] ERROR: cannot parse auth properties ("
+ + (authArg != null ? "--healthCheckAuth" :
ENV_HEALTH_CHECK_AUTH)
+ + "): "
+ + e.getMessage());
+ return EXIT_ERROR;
+ }
+
+ MetricRegistryImpl registry = new
MetricRegistryImpl(Collections.emptyList());
+ ClientMetricGroup metricGroup = new ClientMetricGroup(registry,
"readiness-probe");
+ try (RpcClient rpcClient = RpcClient.create(conf, metricGroup)) {
+ TabletServerGateway gateway =
+ GatewayClientProxy.createGatewayProxy(
+ () -> tabletServer, rpcClient,
TabletServerGateway.class);
+ GetClusterHealthResponse resp =
+ gateway.getClusterHealth(new GetClusterHealthRequest())
+ .get(timeoutMs, TimeUnit.MILLISECONDS);
+ return evaluate(resp);
+ } catch (TimeoutException e) {
+ System.err.println(
+ "[readiness-check] Timeout calling getClusterHealth on "
+ + address
+ + ", treating as not ready");
+ return EXIT_NOT_READY;
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (ExceptionUtils.findThrowable(cause,
UnsupportedVersionException.class)
+ .isPresent()) {
+ System.err.println("[readiness-check] API unsupported: " +
cause.getMessage());
+ return EXIT_API_UNSUPPORTED;
+ }
+ System.err.println(
+ "[readiness-check] Cannot reach tablet server at "
+ + address
+ + ", treating as not ready: "
+ + cause.getMessage());
+ return EXIT_NOT_READY;
+ } catch (Exception e) {
+ System.err.println(
+ "[readiness-check] Cannot reach tablet server at "
+ + address
+ + ", treating as not ready: "
+ + e.getMessage());
+ return EXIT_NOT_READY;
+ } finally {
+ // Best-effort: shut down the metric registry so the probe JVM can
exit cleanly.
+ try {
+ registry.closeAsync();
+ } catch (Exception ignored) {
+ // ignore — probe process is about to exit anyway
+ }
+ }
+ }
+
+ /** Returns the first argument that is non-null and non-blank; otherwise
{@code fallback}. */
+ private static String firstNonBlank(String... candidates) {
+ for (int i = 0; i < candidates.length - 1; i++) {
+ String c = candidates[i];
+ if (c != null && !c.trim().isEmpty()) {
+ return c;
+ }
+ }
+ return candidates[candidates.length - 1];
+ }
+
+ /**
+ * Build the {@link Configuration} fed to {@link RpcClient#create} from
the optional auth
+ * configuration string. Format: semicolon-separated {@code key:value}
pairs, e.g. {@code
+ * client.security.protocol:SASL;client.sasl.mechanism:PLAIN}.
+ */
+ private static Configuration buildConfiguration(String authString) {
+ Configuration conf = new Configuration();
+ if (authString == null || authString.trim().isEmpty()) {
+ return conf;
+ }
+ for (String pair : authString.split(";")) {
+ String trimmed = pair.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ int idx = trimmed.indexOf(':');
+ if (idx <= 0 || idx == trimmed.length() - 1) {
+ throw new IllegalArgumentException(
+ "expected 'key:value' pairs separated by ';', got: \""
+ pair + "\"");
+ }
+ String key = trimmed.substring(0, idx).trim();
+ String value = trimmed.substring(idx + 1).trim();
+ conf.setString(key, value);
+ }
+ return conf;
+ }
+
+ private static int evaluate(GetClusterHealthResponse resp) {
+ int status = resp.getStatus();
+ System.out.println(
+ "[readiness-check] status="
+ + status
+ + " numReplicas="
+ + resp.getNumReplicas()
+ + " inSyncReplicas="
+ + resp.getInSyncReplicas()
+ + " numLeaderReplicas="
+ + resp.getNumLeaderReplicas()
+ + " activeLeaderReplicas="
+ + resp.getActiveLeaderReplicas());
+ return status == STATUS_GREEN ? EXIT_READY : EXIT_NOT_READY;
+ }
+}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ClusterHealthTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ClusterHealthTest.java
new file mode 100644
index 000000000..cf5d153f9
--- /dev/null
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/ClusterHealthTest.java
@@ -0,0 +1,346 @@
+/*
+ * 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 org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.cluster.ServerType;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePartition;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
+import org.apache.fluss.server.metadata.ServerInfo;
+import org.apache.fluss.server.zk.ZkEpoch;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link CoordinatorService#computeClusterHealth}. */
+class ClusterHealthTest {
+
+ private CoordinatorContext ctx;
+
+ @BeforeEach
+ void setUp() {
+ ctx = new CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ ctx.setLiveTabletServers(
+ Arrays.asList(makeServerInfo(0), makeServerInfo(1),
makeServerInfo(2)));
+ }
+
+ @Test
+ void testEmptyCluster() {
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(0);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(0);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(0);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(0);
+ assertThat(resp.getStatus()).isEqualTo(0 /* GREEN */);
+ }
+
+ @Test
+ void testGreenAllInSyncAllLeadersActive() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 2));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1, 2),
Collections.emptyList(), 0, 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(3);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(3);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getStatus()).isEqualTo(0 /* GREEN */);
+ }
+
+ @Test
+ void testYellowIsrIncompleteButLeadersActive() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 2));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(3);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(2);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getStatus()).isEqualTo(1 /* YELLOW */);
+ }
+
+ @Test
+ void testRedLeaderInactive() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 2));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1, 2),
Collections.emptyList(), 0, 1));
+ ctx.addPendingLeaderActivation(tb);
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(3);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(3);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(0);
+ assertThat(resp.getStatus()).isEqualTo(2 /* RED */);
+ }
+
+ @Test
+ void testRedNoLeader() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb,
+ new LeaderAndIsr(
+ LeaderAndIsr.NO_LEADER,
+ 1,
+ Arrays.asList(0, 1),
+ Collections.emptyList(),
+ 0,
+ 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(0);
+ assertThat(resp.getStatus()).isEqualTo(2 /* RED */);
+ }
+
+ @Test
+ void testRedLeaderOnDeadServer() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 5));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(5, 1, Arrays.asList(0, 1, 5),
Collections.emptyList(), 0, 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(0);
+ assertThat(resp.getStatus()).isEqualTo(2 /* RED */);
+ }
+
+ @Test
+ void testNoLeaderAndIsrCountsAsRedAndZeroIsr() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(2);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(0);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(0);
+ assertThat(resp.getStatus()).isEqualTo(2 /* RED */);
+ }
+
+ @Test
+ void testTransitionRedToGreenAfterMarkActive() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+ ctx.addPendingLeaderActivation(tb);
+
+
assertThat(CoordinatorService.computeClusterHealth(ctx).getStatus()).isEqualTo(2
/* RED */);
+
+ ctx.clearPendingLeaderActivation(tb);
+
+ assertThat(CoordinatorService.computeClusterHealth(ctx).getStatus())
+ .isEqualTo(0 /* GREEN */);
+ }
+
+ @Test
+ void testMultipleBucketsMixed() {
+ TableBucket tb1 = new TableBucket(1L, 0);
+ TableBucket tb2 = new TableBucket(1L, 1);
+ TableBucket tb3 = new TableBucket(2L, 0);
+
+ ctx.updateBucketReplicaAssignment(tb1, Arrays.asList(0, 1));
+ ctx.updateBucketReplicaAssignment(tb2, Arrays.asList(1, 2));
+ ctx.updateBucketReplicaAssignment(tb3, Arrays.asList(0, 2));
+
+ ctx.putBucketLeaderAndIsr(
+ tb1, new LeaderAndIsr(0, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb2,
+ new LeaderAndIsr(
+ 1, 1, Collections.singletonList(1),
Collections.emptyList(), 0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb3, new LeaderAndIsr(0, 1, Arrays.asList(0, 2),
Collections.emptyList(), 0, 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(6);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(5);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(3);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(3);
+ assertThat(resp.getStatus()).isEqualTo(1 /* YELLOW */);
+ }
+
+ @Test
+ void testPartitionedTableBucket() {
+ TableBucket tb = new TableBucket(1L, 100L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb,
+ new LeaderAndIsr(
+ 0, 1, Collections.singletonList(0),
Collections.emptyList(), 0, 1));
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(2);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(1);
+ assertThat(resp.getNumLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getStatus()).isEqualTo(1 /* YELLOW */);
+ }
+
+ @Test
+ void testInactiveLeaderClearedOnTableRemoval() {
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+ ctx.addPendingLeaderActivation(tb);
+
+ assertThat(ctx.getPendingLeaderActivationBuckets()).contains(tb);
+
+ ctx.removeTable(1L);
+
+ assertThat(ctx.getPendingLeaderActivationBuckets()).doesNotContain(tb);
+ }
+
+ @Test
+ void testRedWithInactiveLeaderAndIncompleteIsr() {
+ TableBucket tb1 = new TableBucket(1L, 0);
+ TableBucket tb2 = new TableBucket(1L, 1);
+
+ ctx.updateBucketReplicaAssignment(tb1, Arrays.asList(0, 1));
+ ctx.updateBucketReplicaAssignment(tb2, Arrays.asList(0, 1));
+
+ ctx.putBucketLeaderAndIsr(
+ tb1,
+ new LeaderAndIsr(
+ 0, 1, Collections.singletonList(0),
Collections.emptyList(), 0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb2, new LeaderAndIsr(1, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+
+ ctx.addPendingLeaderActivation(tb2);
+
+ GetClusterHealthResponse resp =
CoordinatorService.computeClusterHealth(ctx);
+
+ assertThat(resp.getNumReplicas()).isEqualTo(4);
+ assertThat(resp.getInSyncReplicas()).isEqualTo(3);
+ assertThat(resp.getActiveLeaderReplicas()).isEqualTo(1);
+ assertThat(resp.getStatus()).isEqualTo(2 /* RED */);
+ }
+
+ @Test
+ void testInactiveLeaderClearedOnPartitionRemoval() {
+ TableBucket tb = new TableBucket(1L, 100L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+ ctx.addPendingLeaderActivation(tb);
+
+ assertThat(ctx.getPendingLeaderActivationBuckets()).contains(tb);
+
+ ctx.removePartition(new TablePartition(1L, 100L));
+
+ assertThat(ctx.getPendingLeaderActivationBuckets()).doesNotContain(tb);
+ }
+
+ @Test
+ void testFollowerNotificationDoesNotMarkInactive() {
+ // Simulate sendNotifyLeaderAndIsrRequest: when NotifyLeaderAndIsr is
sent
+ // to a server for a bucket where it is a follower, the bucket should
NOT
+ // be marked inactive. Only the server that is the leader gets marked.
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1),
Collections.emptyList(), 0, 1));
+
+ // Sending to server 1 (follower): leader=0, serverId=1
+ // Follower does not match leader, so addPendingLeaderActivation is
NOT called.
+ assertThat(ctx.isLeaderActive(tb)).isTrue();
+ assertThat(CoordinatorService.computeClusterHealth(ctx).getStatus())
+ .isEqualTo(0 /* GREEN */);
+
+ // Sending to server 0 (leader): leader=0, serverId=0 — matches, so
mark pending.
+ ctx.addPendingLeaderActivation(tb);
+ assertThat(ctx.isLeaderActive(tb)).isFalse();
+
assertThat(CoordinatorService.computeClusterHealth(ctx).getStatus()).isEqualTo(2
/* RED */);
+ }
+
+ @Test
+ void testLeaderChangedBetweenSendAndResponseStaysInactive() {
+ // Simulate processNotifyLeaderAndIsrResponseReceivedEvent: if the
leader
+ // changed between send and response, the responding server is no
longer
+ // the leader, so the bucket must stay inactive.
+ TableBucket tb = new TableBucket(1L, 0);
+ ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1, 2));
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1, 2),
Collections.emptyList(), 0, 1));
+ ctx.addPendingLeaderActivation(tb);
+
+ // Simulate: leader changed from server 0 to server 1 before server 0
responds
+ ctx.putBucketLeaderAndIsr(
+ tb, new LeaderAndIsr(1, 2, Arrays.asList(0, 1, 2),
Collections.emptyList(), 0, 1));
+
+ // Server 0 responds successfully — but it's no longer the leader
+ int respondingServerId = 0;
+ ctx.getBucketLeaderAndIsr(tb)
+ .ifPresent(
+ lai -> {
+ if (lai.leader() == respondingServerId) {
+ ctx.clearPendingLeaderActivation(tb);
+ }
+ });
+
+ // Bucket must stay inactive because the responding server (0) !=
current leader (1)
+ assertThat(ctx.isLeaderActive(tb)).isFalse();
+
assertThat(CoordinatorService.computeClusterHealth(ctx).getStatus()).isEqualTo(2
/* RED */);
+
+ // Now server 1 (the actual leader) responds → bucket becomes active
+ int actualLeaderServerId = 1;
+ ctx.getBucketLeaderAndIsr(tb)
+ .ifPresent(
+ lai -> {
+ if (lai.leader() == actualLeaderServerId) {
+ ctx.clearPendingLeaderActivation(tb);
+ }
+ });
+
+ assertThat(ctx.isLeaderActive(tb)).isTrue();
+ assertThat(CoordinatorService.computeClusterHealth(ctx).getStatus())
+ .isEqualTo(0 /* GREEN */);
+ }
+
+ private static ServerInfo makeServerInfo(int id) {
+ return new ServerInfo(
+ id,
+ "RACK" + id,
+ Endpoint.fromListenersString("CLIENT://host" + id + ":9124"),
+ ServerType.TABLET_SERVER);
+ }
+}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
index 2b788bdb4..b69fe5fa3 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorContextTest.java
@@ -17,17 +17,26 @@
package org.apache.fluss.server.coordinator;
+import org.apache.fluss.cluster.Endpoint;
+import org.apache.fluss.cluster.ServerType;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TableInfo;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.server.metadata.ServerInfo;
import org.apache.fluss.server.zk.ZkEpoch;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
import org.apache.fluss.types.DataTypes;
import org.junit.jupiter.api.Test;
import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
@@ -69,6 +78,124 @@ class CoordinatorContextTest {
assertThat(context.getLakeTableCount()).isEqualTo(2);
}
+ // ---- Pending Leader Activation Tracking Tests ----
+
+ @Test
+ void testAddPendingLeaderActivation() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb1 = new TableBucket(1L, 0);
+ TableBucket tb2 = new TableBucket(1L, 1);
+
+ assertThat(context.getPendingLeaderActivationBuckets()).isEmpty();
+
+ context.addPendingLeaderActivation(tb1);
+ context.addPendingLeaderActivation(tb2);
+
+
assertThat(context.getPendingLeaderActivationBuckets()).containsExactlyInAnyOrder(tb1,
tb2);
+ }
+
+ @Test
+ void testClearPendingLeaderActivation() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb1 = new TableBucket(1L, 0);
+ TableBucket tb2 = new TableBucket(1L, 1);
+
+ context.addPendingLeaderActivations(Arrays.asList(tb1, tb2));
+
+ context.clearPendingLeaderActivation(tb1);
+
assertThat(context.getPendingLeaderActivationBuckets()).containsExactly(tb2);
+
+ context.clearPendingLeaderActivation(tb2);
+ assertThat(context.getPendingLeaderActivationBuckets()).isEmpty();
+ }
+
+ @Test
+ void testClearPendingLeaderActivationForNonExistentBucket() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb1 = new TableBucket(1L, 0);
+
+ // Should not throw
+ context.clearPendingLeaderActivation(tb1);
+ assertThat(context.getPendingLeaderActivationBuckets()).isEmpty();
+ }
+
+ @Test
+ void testRemoveFromPendingLeaderActivations() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb1 = new TableBucket(1L, 0);
+ TableBucket tb2 = new TableBucket(1L, 1);
+ TableBucket tb3 = new TableBucket(2L, 0);
+
+ context.addPendingLeaderActivations(Arrays.asList(tb1, tb2, tb3));
+
+ Set<TableBucket> toRemove = new HashSet<>(Arrays.asList(tb1, tb3));
+ context.removeFromPendingLeaderActivations(toRemove);
+
+
assertThat(context.getPendingLeaderActivationBuckets()).containsExactly(tb2);
+ }
+
+ @Test
+ void testGetPendingLeaderActivationBucketsReturnsUnmodifiableSet() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb1 = new TableBucket(1L, 0);
+ context.addPendingLeaderActivation(tb1);
+
+ Set<TableBucket> pending = context.getPendingLeaderActivationBuckets();
+ assertThat(pending).isUnmodifiable();
+ }
+
+ @Test
+ void testIsLeaderActiveSingleSourceOfTruth() {
+ CoordinatorContext context = new
CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ TableBucket tb = new TableBucket(1L, 0);
+
+ // No LeaderAndIsr → not active
+ assertThat(context.isLeaderActive(tb)).isFalse();
+
+ // Set up live server and LeaderAndIsr
+ context.setLiveTabletServers(
+ Collections.singletonList(
+ new ServerInfo(
+ 0,
+ "RACK0",
+
Endpoint.fromListenersString("CLIENT://host0:9124"),
+ ServerType.TABLET_SERVER)));
+ context.putBucketLeaderAndIsr(
+ tb,
+ new LeaderAndIsr(
+ 0, 1, Collections.singletonList(0),
Collections.emptyList(), 0, 1));
+
+ // Active leader on live server, not pending → active
+ assertThat(context.isLeaderActive(tb)).isTrue();
+
+ // Add to pending → not active
+ context.addPendingLeaderActivation(tb);
+ assertThat(context.isLeaderActive(tb)).isFalse();
+
+ // Clear pending → active again
+ context.clearPendingLeaderActivation(tb);
+ assertThat(context.isLeaderActive(tb)).isTrue();
+
+ // Leader is NO_LEADER → not active
+ context.putBucketLeaderAndIsr(
+ tb,
+ new LeaderAndIsr(
+ LeaderAndIsr.NO_LEADER,
+ 1,
+ Collections.singletonList(0),
+ Collections.emptyList(),
+ 0,
+ 1));
+ assertThat(context.isLeaderActive(tb)).isFalse();
+
+ // Leader on dead server → not active
+ context.putBucketLeaderAndIsr(
+ tb,
+ new LeaderAndIsr(
+ 99, 1, Collections.singletonList(99),
Collections.emptyList(), 0, 1));
+ assertThat(context.isLeaderActive(tb)).isFalse();
+ }
+
private TableInfo createTableInfo(long tableId, TablePath tablePath,
boolean isLake) {
TableDescriptor tableDescriptor =
TableDescriptor.builder()
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatchTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatchTest.java
new file mode 100644
index 000000000..0b3892a1c
--- /dev/null
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatchTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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 org.apache.fluss.exception.NetworkException;
+import org.apache.fluss.metadata.PhysicalTablePath;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrRequest;
+import org.apache.fluss.rpc.messages.NotifyLeaderAndIsrResponse;
+import org.apache.fluss.server.coordinator.event.AccessContextEvent;
+import org.apache.fluss.server.coordinator.event.EventManager;
+import org.apache.fluss.server.zk.ZkEpoch;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.function.BiConsumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for the {@link CoordinatorRequestBatch}. */
+class CoordinatorRequestBatchTest {
+
+ private CoordinatorContext coordinatorContext;
+
+ @BeforeEach
+ void beforeEach() {
+ coordinatorContext = new CoordinatorContext(ZkEpoch.INITIAL_EPOCH);
+ }
+
+ /**
+ * When the NotifyLeaderAndIsr request to the actual leader (leader ==
serverId) fails, the
+ * pending leader activation entry that this request added must be cleared
via an
+ * AccessContextEvent dispatched on the event manager.
+ */
+ @Test
+ void testNotifyLeaderAndIsrSendFailureClearsLeaderPending() {
+ long tableId = 100L;
+ TableBucket tb = new TableBucket(tableId, 0);
+ TablePath tablePath = TablePath.of("db1", "t1");
+
+ coordinatorContext.putTablePath(tableId, tablePath);
+ coordinatorContext.setLiveTabletServers(
+
CoordinatorTestUtils.createServers(Collections.singletonList(0)));
+ coordinatorContext.updateBucketReplicaAssignment(tb,
Collections.singletonList(0));
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(0, 0, Collections.singletonList(0),
Collections.emptyList(), 0, 0);
+ coordinatorContext.putBucketLeaderAndIsr(tb, leaderAndIsr);
+
+ TestCoordinatorChannelManager failingChannel =
newAlwaysFailingChannelManager();
+ EventManager eventManager = newSynchronousAccessContextEventManager();
+
+ CoordinatorRequestBatch batch =
+ new CoordinatorRequestBatch(failingChannel, eventManager,
coordinatorContext);
+ // server 0 IS the leader, so this request will mark tb pending.
+ batch.addNotifyLeaderRequestForTabletServers(
+ Collections.singleton(0),
+ PhysicalTablePath.of(tablePath),
+ tb,
+ Collections.singletonList(0),
+ leaderAndIsr);
+
+ batch.sendRequestToTabletServers(0);
+
+ // The failure callback must clear the pending entry that this request
added.
+
assertThat(coordinatorContext.getPendingLeaderActivationBuckets()).isEmpty();
+ }
+
+ /**
+ * When the NotifyLeaderAndIsr request to a follower (leader != serverId)
fails, the failure
+ * callback must NOT clear any pending leader activation entry.
Specifically, it must not remove
+ * entries added by another in-flight sender targeting the actual leader.
+ */
+ @Test
+ void testNotifyLeaderAndIsrSendFailureToFollowerDoesNotClearOtherPending()
{
+ long tableId = 200L;
+ TableBucket followerTb = new TableBucket(tableId, 0);
+ TableBucket otherLeaderTb = new TableBucket(tableId, 1);
+ TablePath tablePath = TablePath.of("db1", "t2");
+
+ coordinatorContext.putTablePath(tableId, tablePath);
+ coordinatorContext.setLiveTabletServers(
+ CoordinatorTestUtils.createServers(Arrays.asList(0, 1)));
+ coordinatorContext.updateBucketReplicaAssignment(followerTb,
Arrays.asList(0, 1));
+ coordinatorContext.updateBucketReplicaAssignment(otherLeaderTb,
Arrays.asList(0, 1));
+ LeaderAndIsr leaderAndIsr =
+ new LeaderAndIsr(0, 0, Arrays.asList(0, 1),
Collections.emptyList(), 0, 0);
+ coordinatorContext.putBucketLeaderAndIsr(followerTb, leaderAndIsr);
+
+ // Simulate another in-flight sender having already marked
otherLeaderTb pending - we
+ // must not touch this entry on the unrelated follower-send failure.
+ coordinatorContext.addPendingLeaderActivation(otherLeaderTb);
+
+ TestCoordinatorChannelManager failingChannel =
newAlwaysFailingChannelManager();
+ EventManager eventManager = newSynchronousAccessContextEventManager();
+
+ CoordinatorRequestBatch batch =
+ new CoordinatorRequestBatch(failingChannel, eventManager,
coordinatorContext);
+ // Send to server 1, which is a follower for followerTb (leader is 0).
Because
+ // leader != serverId, this request does NOT add followerTb to
pending, so the failure
+ // callback must short-circuit without dispatching an
AccessContextEvent.
+ batch.addNotifyLeaderRequestForTabletServers(
+ Collections.singleton(1),
+ PhysicalTablePath.of(tablePath),
+ followerTb,
+ Arrays.asList(0, 1),
+ leaderAndIsr);
+
+ batch.sendRequestToTabletServers(0);
+
+ // The unrelated leader's pending entry must remain intact.
+ assertThat(coordinatorContext.getPendingLeaderActivationBuckets())
+ .containsExactly(otherLeaderTb);
+ }
+
+ private static TestCoordinatorChannelManager
newAlwaysFailingChannelManager() {
+ return new TestCoordinatorChannelManager() {
+ @Override
+ public void sendBucketLeaderAndIsrRequest(
+ int receiveServerId,
+ NotifyLeaderAndIsrRequest notifyLeaderAndIsrRequest,
+ BiConsumer<NotifyLeaderAndIsrResponse, ? super Throwable>
responseConsumer) {
+ responseConsumer.accept(
+ null, new NetworkException("simulated send failure for
test"));
+ }
+ };
+ }
+
+ /**
+ * Returns an EventManager that synchronously executes {@link
AccessContextEvent}s against the
+ * test's CoordinatorContext, mimicking the serial event-thread semantics.
+ */
+ private EventManager newSynchronousAccessContextEventManager() {
+ return event -> {
+ if (event instanceof AccessContextEvent) {
+ AccessContextEvent<?> accessContextEvent =
(AccessContextEvent<?>) event;
+
accessContextEvent.getAccessFunction().apply(coordinatorContext);
+ }
+ };
+ }
+}
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
index 98e295f9a..8530ad4e7 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TestCoordinatorGateway.java
@@ -70,6 +70,8 @@ import org.apache.fluss.rpc.messages.DropPartitionRequest;
import org.apache.fluss.rpc.messages.DropPartitionResponse;
import org.apache.fluss.rpc.messages.DropTableRequest;
import org.apache.fluss.rpc.messages.DropTableResponse;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
import org.apache.fluss.rpc.messages.GetDatabaseInfoResponse;
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenRequest;
@@ -461,6 +463,12 @@ public class TestCoordinatorGateway implements
CoordinatorGateway {
throw new UnsupportedOperationException();
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public CompletableFuture<RegisterProducerOffsetsResponse>
registerProducerOffsets(
RegisterProducerOffsetsRequest request) {
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java
b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java
index e54987f13..c78270ea5 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/tablet/TestTabletServerGateway.java
@@ -30,6 +30,8 @@ import
org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
import org.apache.fluss.rpc.messages.FetchLogRequest;
import org.apache.fluss.rpc.messages.FetchLogResponse;
+import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
+import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
import org.apache.fluss.rpc.messages.GetDatabaseInfoResponse;
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenRequest;
@@ -234,6 +236,12 @@ public class TestTabletServerGateway implements
TabletServerGateway {
throw new UnsupportedOperationException();
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public CompletableFuture<DatabaseExistsResponse>
databaseExists(DatabaseExistsRequest request) {
throw new UnsupportedOperationException();
diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml
index db770f56c..d5e91d302 100644
--- a/fluss-test-coverage/pom.xml
+++ b/fluss-test-coverage/pom.xml
@@ -448,7 +448,9 @@
<exclude>org.apache.fluss.predicate.*</exclude>
<!-- end exclude for lake source -->
<exclude>org.apache.fluss.lake.source.*</exclude>
- <!-- exclude for dummy class -->
+ <!-- exclude for readiness-probe CLI
tool and dist dummy class -->
+
<exclude>org.apache.fluss.server.tools.ClusterHealthReadinessCheck</exclude>
+
<exclude>org.apache.fluss.server.tools.ClusterHealthReadinessCheck.*</exclude>
<exclude>org.apache.fluss.dist.DummyClass</exclude>
<exclude>org.apache.fluss.flink.DummyClass120</exclude>
<exclude>org.apache.fluss.lake.batch.ArrowRecordBatch</exclude>
diff --git a/helm/templates/sts-coordinator.yaml
b/helm/templates/sts-coordinator.yaml
index 918bf2d06..1d4dd3e9e 100644
--- a/helm/templates/sts-coordinator.yaml
+++ b/helm/templates/sts-coordinator.yaml
@@ -25,6 +25,8 @@ metadata:
spec:
serviceName: coordinator-server-hs
replicas: {{ .Values.coordinator.numberOfReplicas }}
+ updateStrategy:
+ type: RollingUpdate
selector:
matchLabels:
{{- include "fluss.selectorLabels" . | nindent 6 }}
@@ -122,7 +124,7 @@ spec:
echo "bind.listeners: ${BIND_LISTENERS}" >>
$FLUSS_HOME/conf/server.yaml && \
echo "advertised.listeners: ${ADVERTISED_LISTENERS}" >>
$FLUSS_HOME/conf/server.yaml && \
- bin/coordinator-server.sh start-foreground
+ exec bin/coordinator-server.sh start-foreground
livenessProbe:
failureThreshold: 100
timeoutSeconds: 1
diff --git a/helm/templates/sts-tablet.yaml b/helm/templates/sts-tablet.yaml
index 6232544de..200a4cdc7 100644
--- a/helm/templates/sts-tablet.yaml
+++ b/helm/templates/sts-tablet.yaml
@@ -25,6 +25,8 @@ metadata:
spec:
serviceName: tablet-server-hs
replicas: {{ .Values.tablet.numberOfReplicas }}
+ updateStrategy:
+ type: RollingUpdate
selector:
matchLabels:
{{- include "fluss.selectorLabels" . | nindent 6 }}
@@ -119,7 +121,7 @@ spec:
echo "bind.listeners: ${BIND_LISTENERS}" >>
$FLUSS_HOME/conf/server.yaml && \
echo "advertised.listeners: ${ADVERTISED_LISTENERS}" >>
$FLUSS_HOME/conf/server.yaml && \
- bin/tablet-server.sh start-foreground
+ exec bin/tablet-server.sh start-foreground
livenessProbe:
failureThreshold: 100
timeoutSeconds: 1
@@ -128,12 +130,28 @@ spec:
tcpSocket:
port: {{ .Values.listeners.client.port }}
readinessProbe:
- failureThreshold: 100
- timeoutSeconds: 1
- initialDelaySeconds: 10
- periodSeconds: 3
- tcpSocket:
- port: {{ .Values.listeners.client.port }}
+ failureThreshold: {{
.Values.tablet.readinessProbe.failureThreshold | default 360 }}
+ timeoutSeconds: {{ .Values.tablet.readinessProbe.timeoutSeconds |
default 10 }}
+ initialDelaySeconds: {{
.Values.tablet.readinessProbe.initialDelaySeconds | default 15 }}
+ periodSeconds: {{ .Values.tablet.readinessProbe.periodSeconds |
default 5 }}
+ exec:
+ command:
+ - /bin/bash
+ - -c
+ - |
+ export FLUSS_SERVER_ID=${POD_NAME##*-}
+ export READINESS_TIMEOUT_MS={{
.Values.tablet.readinessProbe.rpcTimeoutMs | default 5000 }}
+ export READINESS_TCP_PORT={{ .Values.listeners.client.port }}
+ # bind.listeners binds the tablet's CLIENT endpoint to
${POD_IP},
+ # not 0.0.0.0, so we must probe the pod's own IP instead of
+ # 127.0.0.1. The local tablet then forwards getClusterHealth
+ # to coordinator via the internal listener.
+ export READINESS_TCP_HOST=${POD_IP}
+ export READINESS_HEALTH_CHECK_TIMEOUT_SECONDS={{
.Values.tablet.readinessProbe.healthCheckTimeoutSeconds | default 1200 }}
+ {{- with .Values.tablet.readinessProbe.healthCheckAuth }}
+ export READINESS_HEALTH_CHECK_AUTH={{ . | quote }}
+ {{- end }}
+ exec $FLUSS_HOME/bin/readiness-check.sh
resources:
{{- toYaml .Values.resources.tabletServer | nindent 12 }}
volumeMounts:
diff --git a/helm/values.yaml b/helm/values.yaml
index e14877d11..dd56df982 100644
--- a/helm/values.yaml
+++ b/helm/values.yaml
@@ -43,6 +43,43 @@ tablet:
enabled: false
size: 1Gi
storageClass:
+ # Readiness probe configuration for rolling upgrade gate.
+ # The probe talks to the LOCAL tablet server (127.0.0.1:client-port) and
+ # asks it to forward a getClusterHealth RPC to the Coordinator over the
+ # internal listener. Pod is marked Ready only when the cluster is GREEN.
+ # If the server does not support the API (older version), the probe
+ # immediately falls back to TCP-only port readiness.
+ readinessProbe:
+ # Timeout in ms for the RPC call.
+ rpcTimeoutMs: 5000
+ # Standard Kubernetes probe parameters (tuned for recovery checks).
+ #
+ # The total time Kubernetes will wait before flipping the pod to NotReady
is
+ # initialDelaySeconds + failureThreshold * periodSeconds
+ # = 15 + 360 * 5 = 1815s (~30 minutes) by default.
+ # IMPORTANT: this window MUST be larger than healthCheckTimeoutSeconds
+ # below, otherwise Kubernetes will mark the pod NotReady BEFORE the
+ # in-script health-check fallback has a chance to latch the TCP fast path,
+ # which would still wedge the rolling upgrade. If you tune
+ # healthCheckTimeoutSeconds upward, raise failureThreshold accordingly so
+ # the inequality (failureThreshold * periodSeconds >
healthCheckTimeoutSeconds)
+ # always holds.
+ failureThreshold: 360
+ timeoutSeconds: 10
+ initialDelaySeconds: 15
+ periodSeconds: 5
+ # Maximum wall-clock seconds the cluster health gate is allowed to keep a
+ # freshly-created tablet pod NotReady on its first boot. After this budget
+ # is exhausted the probe latches the TCP-only fast path so a permanently-
+ # unrecoverable cluster cannot wedge a rolling upgrade indefinitely.
+ # Default: 1200 (20 minutes).
+ healthCheckTimeoutSeconds: 1200
+ # Optional client auth configuration passed to the probe's RpcClient as a
+ # single-line string of semicolon-separated `key:value` pairs (no quoting
+ # or escaping needed). Required only when the local client listener
+ # enforces SASL. Example:
+ # healthCheckAuth:
"client.security.protocol:SASL;client.sasl.mechanism:PLAIN;client.security.sasl.username:admin;client.security.sasl.password:admin-pass"
+ healthCheckAuth: ""
extraVolumes: []
extraVolumeMounts: []
initContainers: []
diff --git a/website/docs/install-deploy/deploying-with-helm.md
b/website/docs/install-deploy/deploying-with-helm.md
index 5bb05dba7..949ff92b1 100644
--- a/website/docs/install-deploy/deploying-with-helm.md
+++ b/website/docs/install-deploy/deploying-with-helm.md
@@ -771,6 +771,43 @@ helm upgrade fluss ./helm -f values-new.yaml
The StatefulSets support rolling updates. When you update the configuration,
pods will be restarted one by one to maintain availability.
+#### Cluster Health Readiness Probe
+
+The TabletServer readiness probe performs a two-step check to ensure safe
rolling upgrades:
+
+1. **Local TCP port check** — confirms the TabletServer process is up and
listening.
+2. **Cluster health check** — calls the Coordinator's Cluster Health API to
confirm cluster status is GREEN (all replicas
+ in-sync, all leaders active) across the cluster.
+
+Only when both steps pass is the pod marked as Ready, allowing the StatefulSet
controller to proceed to the next pod.
+If the Coordinator does not support the Cluster Health API (e.g., during a
mixed-version upgrade from an older version),
+the probe immediately falls back to TCP-only port readiness.
+
+You can tune the probe parameters in your values:
+
+```yaml
+tablet:
+ readinessProbe:
+ # Timeout in ms for the RPC call to Coordinator (default: 5000)
+ rpcTimeoutMs: 5000
+ # Standard Kubernetes probe parameters (tuned for recovery checks)
+ failureThreshold: 200
+ timeoutSeconds: 10
+ initialDelaySeconds: 15
+ periodSeconds: 5
+```
+
+| Parameter | Default | Description
|
+|--------------------|---------|------------------------------------------------------------------------------------------------------------------------------------------------|
+| `rpcTimeoutMs` | `5000` | Timeout in milliseconds for the RPC call to
the Coordinator.
|
+| `failureThreshold` | `200` | Max consecutive probe failures before marking
the pod as unready. With `periodSeconds=5`, this allows up to ~16 minutes for
recovery. |
+| `periodSeconds` | `5` | How often the probe runs.
|
+
+:::note
+The CoordinatorServer does not need the Cluster Health API probe — it does not
host data replicas, so a simple TCP check is sufficient.
+The Coordinator should be upgraded **after** all TabletServers are fully
upgraded and recovered.
+:::
+
## Custom Container Images
### Building Custom Images
@@ -805,7 +842,7 @@ image:
### Health Checks
-The chart includes liveness and readiness probes:
+The chart includes liveness and readiness probes. By default, both use TCP
socket checks:
```yaml
livenessProbe:
@@ -823,6 +860,9 @@ readinessProbe:
failureThreshold: 100
```
+For TabletServers, you can enable the Cluster Health readiness probe for safe
rolling upgrades.
+See [Cluster Health Readiness Probe](#cluster-health-readiness-probe) for
details.
+
### Logs
Access logs from different components:
diff --git a/website/docs/maintenance/observability/monitor-metrics.md
b/website/docs/maintenance/observability/monitor-metrics.md
index 4fcc83687..5383250b8 100644
--- a/website/docs/maintenance/observability/monitor-metrics.md
+++ b/website/docs/maintenance/observability/monitor-metrics.md
@@ -294,8 +294,8 @@ Some metrics might not be exposed when using other JVM
implementations (e.g. IBM
</thead>
<tbody>
<tr>
- <th rowspan="27"><strong>coordinator</strong></th>
- <td style={{textAlign: 'center', verticalAlign: 'middle' }}
rowspan="10">-</td>
+ <th rowspan="28"><strong>coordinator</strong></th>
+ <td style={{textAlign: 'center', verticalAlign: 'middle' }}
rowspan="11">-</td>
<td>activeCoordinatorCount</td>
<td>The number of active CoordinatorServer (only leader) in this
cluster.</td>
<td>Gauge</td>
@@ -345,6 +345,11 @@ Some metrics might not be exposed when using other JVM
implementations (e.g. IBM
<td>The total number of replicas in the progress to be deleted in this
cluster.</td>
<td>Gauge</td>
</tr>
+ <tr>
+ <td>pendingLeaderActivationCount</td>
+ <td>The number of buckets currently waiting for the leader-activation
acknowledgement from the target tablet server. A non-zero, sustained value
blocks the cluster-health API from reporting GREEN and will hold the readiness
gate during a rolling upgrade; if it stays non-zero indefinitely it indicates a
stuck activation that requires investigation.</td>
+ <td>Gauge</td>
+ </tr>
<tr>
<td rowspan="3">event</td>
<td>eventQueueSize</td>
diff --git a/website/docs/maintenance/operations/upgrading.md
b/website/docs/maintenance/operations/upgrading.md
index 2b75557b6..1eb768a46 100644
--- a/website/docs/maintenance/operations/upgrading.md
+++ b/website/docs/maintenance/operations/upgrading.md
@@ -57,6 +57,44 @@ To upgrade the `TabletServers`, follow these steps
one-by-one for each `TabletSe
./fluss-$FLUSS_VERSION$/bin/tablet-server.sh start
```
+**Wait for the cluster to recover before upgrading the next TabletServer**
+
+After restarting a TabletServer, you should wait for the cluster to fully
recover before proceeding to the next one.
+Upgrading too quickly can cause multiple servers to be in an unrecovered state
simultaneously, which may break min-ISR guarantees
+and affect data availability.
+
+You can use the **Cluster Health API** to monitor the cluster's health status:
+
+```java
+Admin admin = connection.getAdmin();
+
+ClusterHealth health = admin.getClusterHealth().get();
+System.out.println("Status: " + health.getStatus());
+System.out.println("Replicas: " + health.getInSyncReplicas() + "/" +
health.getNumReplicas());
+System.out.println("Leaders: " + health.getActiveLeaderReplicas() + "/" +
health.getNumLeaderReplicas());
+
+if (health.getStatus() == ClusterHealthStatus.GREEN) {
+ // Safe to proceed with the next TabletServer
+}
+```
+
+The API returns the following health metrics:
+
+| Field | Meaning |
+|-------|---------|
+| `status` | Cluster health: GREEN (all replicas in-sync, all leaders active),
YELLOW (all leaders active, some followers not in-sync), RED (some leaders not
active) |
+| `numReplicas` | Total number of assigned replicas across all buckets |
+| `inSyncReplicas` | Total number of in-sync replicas across all buckets |
+| `numLeaderReplicas` | Total number of leader slots (one per bucket) |
+| `activeLeaderReplicas` | Number of active leaders (leader alive and
acknowledged) |
+
+Wait until the status is GREEN before upgrading the next TabletServer. GREEN
means all replicas are in-sync and all leaders are active — fully recovered.
+
+:::tip
+For Kubernetes deployments, you can enable the Cluster Health API readiness
probe in the Helm chart to automate this check.
+See [Deploying with Helm Charts — Cluster Health Readiness
Probe](docs/install-deploy/deploying-with-helm.md#cluster-health-readiness-probe)
for details.
+:::
+
### Upgrade the CoordinatorServer
After all `TabletServers` have been upgraded, you can proceed to upgrade the
`CoordinatorServer` by following these steps:
