You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "swuferhong (via GitHub)" <gi...@apache.org> on 2026/05/29 02:03:03 UTC
[PR] [server] Add Cluster Health API implementation [fluss]
swuferhong opened a new pull request, #3400:
URL: https://github.com/apache/fluss/pull/3400
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #3399
<!-- What is the purpose of the change -->
- Add `GetClusterHealth` RPC to Coordinator that computes cluster health from in-memory state
- Track inactive leaders in `CoordinatorContext` (marked inactive on `NotifyLeaderAndIsr` send,
marked active on successful response when responding server is still the leader)
- Handle send failures in `CoordinatorRequestBatch` by synthesizing error responses to clear
pending inactive state
- Add client API `Admin.getClusterHealth()` with `ClusterHealth` / `ClusterHealthStatus` types
- Add `ClusterHealthReadinessCheck` CLI tool in fluss-dist (exit 0=GREEN, 1=not ready, 2=API unsupported)
- Add `readiness-check.sh` two-step readiness probe script (TCP + Cluster Health API)
with first-boot detection and grace period for API-unsupported (mixed-version rolling upgrade)
- Wire tablet-server readiness probe to `readiness-check.sh` in Helm chart
- Add documentation for Helm deployment and upgrade guide
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3384872343
##########
helm/templates/sts-tablet.yaml:
##########
@@ -128,12 +130,21 @@ 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 200 }}
+ 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 }}
+ export READINESS_BOOTSTRAP_SERVERS="coordinator-server-hs.${POD_NAMESPACE}.svc.cluster.local:{{ .Values.listeners.client.port }}"
Review Comment:
Switched the readiness probe's bootstrap server to 127.0.0.1 so it no longer depends on the coordinator's address.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #3400:
URL: https://github.com/apache/fluss/pull/3400#issuecomment-4670468189
@wuchong comments addressed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3386947845
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java:
##########
@@ -409,10 +409,19 @@ private void sendNotifyLeaderAndIsrRequest(int coordinatorEpoch) {
notifyRequestEntry : notifyLeaderAndIsrRequestMap.entrySet()) {
// send request for each tablet server
Integer serverId = notifyRequestEntry.getKey();
+ Set<TableBucket> buckets = notifyRequestEntry.getValue().keySet();
NotifyLeaderAndIsrRequest notifyLeaderAndIsrRequest =
makeNotifyLeaderAndIsrRequest(
coordinatorEpoch, notifyRequestEntry.getValue().values());
+ for (Map.Entry<TableBucket, PbNotifyLeaderAndIsrReqForBucket> entry :
+ notifyRequestEntry.getValue().entrySet()) {
+ int leader = entry.getValue().getLeader();
+ if (leader == serverId || leader == LeaderAndIsr.NO_LEADER) {
Review Comment:
Why should the NO_LEADER state be included in the pending leader set? Once a NO_LEADER bucket is added to this set, it is never cleared upon receiving the NotifyLeaderAndIsrResponse. Since we already treat NO_LEADER buckets as inactive leaders in org.apache.fluss.server.coordinator.CoordinatorContext#isLeaderActive, there is no need to handle the NotifyLeaderAndIsr path for NO_LEADER buckets.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3345479836
##########
helm/templates/sts-coordinator.yaml:
##########
@@ -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
Review Comment:
Without exec, the Fluss process is not PID 1 in the pod, so graceful shutdown signals won't reach it. I fixed this as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on PR #3400:
URL: https://github.com/apache/fluss/pull/3400#issuecomment-4609901083
@loserwang1024 comments addressed. PLTA.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3338499807
##########
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 leader buckets that are currently inactive — a leader is inactive from the moment we
+ * send NotifyLeaderAndIsr until the target server responds successfully confirming it is the
+ * leader. Also inactive when leader == NO_LEADER.
+ */
+ private final Set<TableBucket> inactiveLeaderBuckets = new HashSet<>();
Review Comment:
Naming: inactiveLeaderBuckets → pendingLeaderActivationBuckets
The name over-claims. The set is only populated in sendNotifyLeaderAndIsrRequest — buckets where we've sent a NotifyLeaderAndIsr and await leadership confirmation. Buckets leaderless for other reasons (server offline, rebalance with no new leader yet) aren't in it, though their leader is effectively inactive. So it really means "leader change dispatched, awaiting confirmation" — pendingLeaderActivationBuckets fits. Avoid pendingIsr...: this has nothing to do with ISR (that's a separate dimension driving YELLOW).
Bigger issue: "active leader" is defined twice. computeClusterHealth checks leader != NO_LEADER && liveServers.contains(leader) && !inSet, but the helper only checks the set:
```java
public boolean isLeaderActive(TableBucket bucket) {
return !inactiveLeaderBuckets.contains(bucket); // ignores liveServers / NO_LEADER
}
```
So for a leader on a dead server, isLeaderActive() says true while computeClusterHealth() says RED — inconsistent, and a trap for future reuse.
Suggestion: rename the set + make isLeaderActive the single source of truth:
```java
public boolean isLeaderActive(TableBucket tb) {
return getBucketLeaderAndIsr(tb)
.map(lai -> lai.leader() != LeaderAndIsr.NO_LEADER
&& liveTabletServers.containsKey(lai.leader())
&& !pendingLeaderActivationBuckets.contains(tb))
.orElse(false);
}
```
Then computeClusterHealth just calls ctx.isLeaderActive(tb).
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -291,6 +294,62 @@ public CoordinatorService(
this.coordinatorLeaderElection = coordinatorLeaderElection;
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ 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;
+ Set<TableBucket> inactiveLeaders = ctx.getInactiveLeaderBuckets();
+
+ for (TableBucket tb : ctx.getAllBuckets()) {
+ List<Integer> assignment = ctx.getAssignment(tb);
+ numReplicas += assignment.size();
+ numLeaderReplicas++;
+
+ Optional<LeaderAndIsr> laiOpt = ctx.getBucketLeaderAndIsr(tb);
+ if (laiOpt.isPresent()) {
+ LeaderAndIsr lai = laiOpt.get();
+ inSyncReplicas += lai.isr().size();
+ if (lai.leader() != LeaderAndIsr.NO_LEADER
+ && ctx.getLiveTabletServers().containsKey(lai.leader())
+ && !inactiveLeaders.contains(tb)) {
+ activeLeaderReplicas++;
+ }
+ }
+ }
+
+ // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
+ int status;
+ if (numLeaderReplicas == 0) {
Review Comment:
This can merge with else {
status = 0; // GREEN
}
##########
fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java:
##########
@@ -574,6 +576,18 @@ public CompletableFuture<DescribeClusterConfigsResponse> describeClusterConfigs(
new DescribeClusterConfigsResponse().addAllConfigs(toPbConfigEntries(configs)));
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
Review Comment:
Why not move to CoordinatorGateway? Could tablet server support it?
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -974,10 +975,17 @@ private void processNotifyLeaderAndIsrResponseReceivedEvent(
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.markLeaderActive(tb);
}
}
if (!offlineReplicas.isEmpty()) {
- // trigger replicas to offline
Review Comment:
why remove this comment?
##########
helm/templates/sts-coordinator.yaml:
##########
@@ -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
Review Comment:
why need to change it?
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java:
##########
@@ -422,12 +433,20 @@ private void sendNotifyLeaderAndIsrRequest(int coordinatorEpoch) {
"Failed to send notify leader and isr request to tablet server {}.",
serverId,
throwable);
- // todo: in FLUSS-55886145, we will introduce a sender thread to send
- // the request, and retry if encounter any error; It may happens that
- // the tablet server is offline and will always got error. But,
- // 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.
+ // Treat all buckets as failed — clears pending state and triggers
+ // re-election via onReplicaBecomeOffline.
+ List<NotifyLeaderAndIsrResultForBucket> failedResults =
+ new ArrayList<>();
+ ApiError sendError =
Review Comment:
Currently, after this modification, even a network timeout exception, the CoordinatorEventProcessor#processNotifyLeaderAndIsrResponseReceivedEvent will mark the server as offlineReplicas
won't is any problem?
<img width="1062" height="280" alt="Image" src="https://github.com/user-attachments/assets/ecbd5a3a-4568-40ed-8c5d-d8871bd94fd5" />
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3387486885
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java:
##########
@@ -428,6 +437,13 @@ private void sendNotifyLeaderAndIsrRequest(int coordinatorEpoch) {
// 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.
+ for (TableBucket tb : buckets) {
+ coordinatorContext.clearPendingLeaderActivation(tb);
Review Comment:
Besides, `coordinatorContext.clearPendingLeaderActivation(...)` is called inside the NotifyLeaderAndIsr send-failure callback, which runs on the RPC completion thread (whenComplete is attached with no executor), not the single coordinator event thread. CoordinatorContext is @NotThreadSafe and pendingLeaderActivationBuckets is a plain HashSet that is concurrently read/mutated on the event thread (e.g. addPendingLeaderActivation, isLeaderActive, computeClusterHealth via AccessContextEvent). This is an unsynchronized data race. The success branch right below correctly hands off via eventManager.put(...) instead of touching the context directly — the failure branch should do the same.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java:
##########
@@ -428,6 +437,13 @@ private void sendNotifyLeaderAndIsrRequest(int coordinatorEpoch) {
// 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.
+ for (TableBucket tb : buckets) {
+ coordinatorContext.clearPendingLeaderActivation(tb);
Review Comment:
When a NotifyLeaderAndIsr request to a follower fails, buckets still contains the bucket whose pending activation was added for the actual leader. Clearing every bucket here can remove that leader's pending state before the leader has acknowledged the request, allowing getClusterHealth() to report GREEN during a rolling upgrade while the new leader is not confirmed active.
##########
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<>();
Review Comment:
Suggest adding a gauge metric for `pendingLeaderActivationBuckets`.
This set is the core signal behind the RED status of the health API: `isLeaderActive` returns false for any bucket in it, which makes the health API report RED and blocks the rolling upgrade at the readiness gate. If a leader bucket's activation never gets cleared (e.g. due to an exception or a missed clear), the upgrade hangs at NotReady indefinitely, while operators only see a pod stuck NotReady with no way to tell which bucket is stuck or for how long — they'd have to dig through coordinator logs, which is expensive to diagnose.
Suggest registering a gauge in `CoordinatorMetricGroup` (e.g. `pendingLeaderActivationCount`) backed by the existing read-only accessor `coordinatorContext.getPendingLeaderActivationBuckets().size()`. This mirrors the existing `gauge(MetricNames.BUCKET_COUNT, buckets::size)` in the same file and is consistent in spirit with `offlineBucketCount` / `replicasToDeleteCount`. A value that stays non-zero over time then becomes directly alertable/observable on dashboards, turning "upgrade hangs" from a hard-to-debug mystery into a visible curve.
##########
fluss-dist/pom.xml:
##########
@@ -263,8 +263,24 @@
<finalName>${project.artifactId}-${project.version}</finalName>
<artifactSet>
<includes>
- <!-- only include log4j-api to package source and javadoc to make sonatype happy -->
- <include>org.apache.logging.log4j:log4j-api</include>
+ <!--
+ IMPORTANT: do NOT include log4j-api here.
Review Comment:
we should revert this, otherwise, release will fail on sonatype.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3345545972
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java:
##########
@@ -422,12 +433,20 @@ private void sendNotifyLeaderAndIsrRequest(int coordinatorEpoch) {
"Failed to send notify leader and isr request to tablet server {}.",
serverId,
throwable);
- // todo: in FLUSS-55886145, we will introduce a sender thread to send
- // the request, and retry if encounter any error; It may happens that
- // the tablet server is offline and will always got error. But,
- // 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.
+ // Treat all buckets as failed — clears pending state and triggers
+ // re-election via onReplicaBecomeOffline.
+ List<NotifyLeaderAndIsrResultForBucket> failedResults =
+ new ArrayList<>();
+ ApiError sendError =
Review Comment:
Good catch. I have reverted it — the original "just ignore" approach was intentional (the coordinator detects actual server death via heartbeat timeout). The only thing I kept is clearing the pendingLeaderActivationBuckets on send failure, so the health API doesn't report stale `RED` for buckets that are simply stuck in the pending set. Actual failover is still handled by the existing heartbeat/offline mechanism.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3382089833
##########
fluss-dist/src/main/resources/bin/readiness-check.sh:
##########
@@ -0,0 +1,249 @@
+#!/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: query the Coordinator's Cluster Health API
+# 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 Coordinator unreachable)
+# 2 = API unsupported → shell handles grace period then TCP fallback
+#
+# Environment variables (set by helm template or container spec):
+# FLUSS_HOME - Fluss installation directory
+# FLUSS_CONF_DIR - Configuration directory (default: $FLUSS_HOME/conf)
+# READINESS_TIMEOUT_MS - Timeout for Health API call (default: 5000)
+# READINESS_TCP_HOST - Host for TCP check (default: $POD_IP or 127.0.0.1)
+# READINESS_TCP_PORT - Port for TCP check (default: 9124)
+# READINESS_BOOTSTRAP_SERVERS - Coordinator bootstrap address
+# READINESS_GRACE_SECS - Grace period when API is unsupported (default: 60)
+# ==============================================================================
+
+set -o pipefail
+
+# ---- Configuration ----
+
+FLUSS_HOME="${FLUSS_HOME:-/opt/fluss}"
+FLUSS_CONF_DIR="${FLUSS_CONF_DIR:-${FLUSS_HOME}/conf}"
+TIMEOUT_MS="${READINESS_TIMEOUT_MS:-5000}"
+TCP_HOST="${READINESS_TCP_HOST:-${POD_IP:-127.0.0.1}}"
+TCP_PORT="${READINESS_TCP_PORT:-9124}"
+BOOTSTRAP_SERVERS="${READINESS_BOOTSTRAP_SERVERS:-}"
+GRACE_SECS="${READINESS_GRACE_SECS:-60}"
+
+# Marker files for tracking state across probe invocations
+MARKER_DIR="/tmp/fluss-readiness"
+FIRST_READY_MARKER="${MARKER_DIR}/first-ready"
+API_UNSUPPORTED_SINCE="${MARKER_DIR}/api-unsupported-since"
+
+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)
+run_recovery_check() {
+ local conf_dir="$1"
+ local timeout_ms="$2"
+
+ # Construct classpath (same logic as config.sh)
+ local classpath=""
+ local fluss_server_jar=""
+ while IFS= read -r -d '' jarfile; do
+ if [[ "$jarfile" =~ .*/fluss-server[^/]*.jar$ ]]; then
+ fluss_server_jar="$jarfile"
+ elif [[ -z "$classpath" ]]; then
+ classpath="$jarfile"
+ else
+ classpath="${classpath}:${jarfile}"
+ fi
+ done < <(find "${FLUSS_HOME}/lib" ! -type d -name '*.jar' -print0 | sort -z)
+
+ if [[ -n "$fluss_server_jar" ]]; then
+ classpath="${classpath}:${fluss_server_jar}"
+ fi
+
+ if [[ -z "$classpath" ]]; then
+ echo "[readiness-check] ERROR: No jars found in ${FLUSS_HOME}/lib"
+ return 3
+ fi
+
+ # Find Java
+ local java_cmd="java"
+ if [[ -n "${JAVA_HOME}" && -x "${JAVA_HOME}/bin/java" ]]; then
+ java_cmd="${JAVA_HOME}/bin/java"
+ fi
+
+ # Build optional --bootstrapServers argument
+ local bootstrap_arg=""
+ if [[ -n "${BOOTSTRAP_SERVERS}" ]]; then
+ bootstrap_arg="--bootstrapServers ${BOOTSTRAP_SERVERS}"
+ fi
+
+ # Run the check (suppress JVM startup noise, only care about exit code + output)
+ local output
+ output=$("${java_cmd}" \
+ -XX:+IgnoreUnrecognizedVMOptions \
+ -Xmx64m \
+ -classpath "${classpath}" \
+ org.apache.fluss.dist.ClusterHealthReadinessCheck \
+ --configDir "${conf_dir}" \
+ --timeoutMs "${timeout_ms}" \
+ ${bootstrap_arg} 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
+}
+
+# Record when API unsupported was first detected
+mark_api_unsupported() {
+ if [[ ! -f "${API_UNSUPPORTED_SINCE}" ]]; then
+ date +%s > "${API_UNSUPPORTED_SINCE}"
+ fi
+}
+
+# Clear API-unsupported marker
+clear_api_unsupported() {
+ rm -f "${API_UNSUPPORTED_SINCE}"
+}
+
+# Check if grace period for API-unsupported has elapsed
+is_grace_period_elapsed() {
+ if [[ ! -f "${API_UNSUPPORTED_SINCE}" ]]; then
+ return 1 # not elapsed (no marker)
+ fi
+ local since
+ since=$(cat "${API_UNSUPPORTED_SINCE}")
+ local now
+ now=$(date +%s)
+ local elapsed=$(( now - since ))
+ if [[ $elapsed -ge ${GRACE_SECS} ]]; then
+ return 0 # elapsed
+ fi
+ return 1 # not elapsed yet
+}
+
+# ---- 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
+
+# ---- Step 2: Cluster health check ----
+
+# First-boot detection:
+# On the very first startup (no previous data), there's nothing to recover.
+# We use a marker file: once the cluster has been "ready" at least once, we
+# know subsequent probes after a restart are "upgrade" scenarios that need
+# the full recovery gate.
+if [[ ! -f "${FIRST_READY_MARKER}" ]]; then
+ run_recovery_check "${FLUSS_CONF_DIR}" "${TIMEOUT_MS}"
+ local_exit=$?
+
+ case $local_exit in
+ 0)
+ log_to_main "[readiness-check] First boot: GREEN, marking ready"
+ touch "${FIRST_READY_MARKER}"
+ clear_api_unsupported
+ exit 0
+ ;;
+ 1)
+ log_to_main "[readiness-check] First boot: not ready (exit 1)"
+ exit 1
+ ;;
+ 2)
+ log_to_main "[readiness-check] First boot: API unsupported, TCP-only fallback"
+ touch "${FIRST_READY_MARKER}"
+ exit 0
+ ;;
+ *)
+ log_to_main "[readiness-check] Config error, falling back to TCP-only"
+ exit 0
+ ;;
+ esac
+fi
+
+# Upgrade/restart scenario (marker file exists = has been ready before)
+run_recovery_check "${FLUSS_CONF_DIR}" "${TIMEOUT_MS}"
+local_exit=$?
+
+case $local_exit in
+ 0)
+ log_to_main "[readiness-check] Upgrade check: PASSED (exit 0)"
+ clear_api_unsupported
+ exit 0
+ ;;
+ 1)
+ log_to_main "[readiness-check] Upgrade check: BLOCKED (exit 1) — waiting for recovery"
+ clear_api_unsupported
+ exit 1
+ ;;
+ 2)
+ mark_api_unsupported
+ if is_grace_period_elapsed; then
+ log_to_main "[readiness-check] API unsupported for >${GRACE_SECS}s, TCP fallback"
+ exit 0
+ fi
+ log_to_main "[readiness-check] API unsupported, waiting (grace period)"
+ exit 1
+ ;;
+ *)
+ log_to_main "[readiness-check] Config error (exit $local_exit), treating as not ready"
+ exit 1
+ ;;
+esac
Review Comment:
Suggest removing the `READINESS_GRACE_SECS` grace-period logic entirely (this `mark_api_unsupported` / `is_grace_period_elapsed` machinery and the `--default 60` plumbing).
The grace period is meant to handle the mixed-version window during a rolling upgrade — "wait a bit, maybe the Coordinator finishes upgrading and starts supporting the API." But it doesn't actually help, and it has a real cost:
- On the **first upgrade from a version that predates this API**, the whole cluster is on the old Coordinator, so the API returns `exit 2` (unsupported) for the *entire* upgrade. Every tablet-server pod is then forced to sit at NotReady for the full `GRACE_SECS` (default 60s) before falling back to TCP-only. Since the StatefulSet rolls serially, that's `60s × pod count` of pure dead time added to the upgrade, with zero protection gained (the API genuinely cannot be available in this window).
- The case the grace period is *trying* to protect — "Coordinator temporarily unreachable / still electing" — should already be `exit 1` (not ready, retry next cycle), not `exit 2`. A clean `UnsupportedVersionException` means the server is up and definitively does not support the API, so there is nothing to wait for: we should TCP-fallback immediately.
So `exit 2` can just latch the fast path right away, same as the config-error `*` branch below it. This removes the grace-period env var, the two marker helpers, and the first-upgrade slowdown.
##########
helm/templates/sts-tablet.yaml:
##########
@@ -128,12 +130,21 @@ 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 200 }}
+ 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 }}
+ export READINESS_BOOTSTRAP_SERVERS="coordinator-server-hs.${POD_NAMESPACE}.svc.cluster.local:{{ .Values.listeners.client.port }}"
Review Comment:
Will the coordinator address change in scenarios such as upgrades, restarts, or failover migrations? If it does, could this prevent the tablet server from connecting to the coordinator during its next restart or upgrade?
##########
fluss-rpc/src/main/proto/FlussApi.proto:
##########
@@ -1369,4 +1369,14 @@ message PbLiteralValue {
optional int64 timestamp_millis_value = 11; // Epoch millis
optional int32 timestamp_nano_of_millis_value = 12; // Nano of millis
optional bytes decimal_bytes = 13; // Serialized decimal (non-compact mode)
+}
+
+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
Review Comment:
nit: move the RPC messages to before the line of
```
// --------------- Inner classes ----------------
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Add Cluster Health API implementation [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #3400:
URL: https://github.com/apache/fluss/pull/3400
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org