Skip to content

Commit b3d64f5

Browse files
committed
[server] Support DescribeTabletServers API for safe scaling and upgrades
Add a read-only DescribeTabletServers RPC to the Coordinator, following the pattern of GetClusterHealth (#3399). It returns, per tablet server, the same four counters getClusterHealth() reports cluster-wide, scoped to that server: numReplicas, inSyncReplicas, numLeaderReplicas and activeLeaderReplicas. This lets operational tooling (e.g. the Fluss Kubernetes Operator): - gate scale-in on a server being empty (numReplicas == 0), - gate rolling upgrades on a per-server green predicate (inSyncReplicas == numReplicas && activeLeaderReplicas == numLeaderReplicas), - report per-server tablet load in cluster status. The counters are computed from in-memory CoordinatorContext state on the coordinator event thread via AccessContextEvent. Tablet servers forward the request to the coordinator. Live and shutting-down servers are always reported, even with zero replicas, and dead servers that still hold assigned replicas remain visible so a scale-in gate cannot treat them as drained. Closes #3570
1 parent 4c89491 commit b3d64f5

15 files changed

Lines changed: 649 additions & 1 deletion

File tree

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

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -799,6 +799,29 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
799799
*/
800800
CompletableFuture<ClusterHealth> getClusterHealth();
801801

802+
/**
803+
* Describe the replica load of each tablet server in the cluster asynchronously.
804+
*
805+
* <p>The returned list contains {@link TabletServerDescription} counters per tablet server.
806+
* Live tablet servers are always included, even when they host no replicas.
807+
*
808+
* <p>This API is designed for operational tooling (e.g., a Kubernetes operator):
809+
*
810+
* <ul>
811+
* <li>scale-in safety gate - a tablet server may only be removed once its {@code numReplicas}
812+
* is {@code 0}, since terminating a non-empty server causes under-replication or data
813+
* unavailability;
814+
* <li>rolling-upgrade gate - only proceed to the next server when the previous one is green
815+
* again ({@code inSyncReplicas == numReplicas && activeLeaderReplicas ==
816+
* numLeaderReplicas});
817+
* <li>reporting per-server tablet load in cluster status.
818+
* </ul>
819+
*
820+
* @return a {@link CompletableFuture} that completes with the per tablet server replica loads.
821+
* @since 1.0
822+
*/
823+
CompletableFuture<List<TabletServerDescription>> describeTabletServers();
824+
802825
/**
803826
* List per-bucket remote log manifest entries for a table or partition scope.
804827
*

‎fluss-client/src/main/java/org/apache/fluss/client/admin/FlussAdmin.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@
6767
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
6868
import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest;
6969
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
70+
import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
7071
import org.apache.fluss.rpc.messages.DropAclsRequest;
7172
import org.apache.fluss.rpc.messages.DropDatabaseRequest;
7273
import org.apache.fluss.rpc.messages.DropTableRequest;
@@ -942,6 +943,12 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
942943
.thenApply(ClientRpcMessageUtils::toClusterHealth);
943944
}
944945

946+
@Override
947+
public CompletableFuture<List<TabletServerDescription>> describeTabletServers() {
948+
return gateway.describeTabletServers(new DescribeTabletServersRequest())
949+
.thenApply(ClientRpcMessageUtils::toTabletServerDescriptions);
950+
}
951+
945952
@VisibleForTesting
946953
public AdminGateway getAdminGateway() {
947954
return gateway;
Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.fluss.client.admin;
19+
20+
import org.apache.fluss.annotation.PublicEvolving;
21+
22+
import java.util.Objects;
23+
24+
/**
25+
* Per tablet server replica load returned by {@link Admin#describeTabletServers()}.
26+
*
27+
* <p>It reports the replica counters of a single tablet server: how many replicas it hosts, how
28+
* many of those are in sync, how many buckets it leads, and how many of those leaders are active.
29+
* The {@link ClusterHealth} aggregates the same counters across all tablet servers.
30+
*
31+
* <p>Operational tooling can derive from it:
32+
*
33+
* <ul>
34+
* <li>a scale-in safety gate: the server is empty and safe to remove iff {@code numReplicas ==
35+
* 0};
36+
* <li>a per-server health (green) predicate for rolling upgrades: {@code inSyncReplicas ==
37+
* numReplicas && activeLeaderReplicas == numLeaderReplicas}.
38+
* </ul>
39+
*
40+
* <p>Note: a bucket without an elected leader is attributed to no server's {@code
41+
* numLeaderReplicas}, so the per-server values sum to the cluster-wide {@link
42+
* ClusterHealth#getNumLeaderReplicas()} only when every bucket has an elected leader. Such buckets
43+
* still surface through {@code inSyncReplicas < numReplicas} on the assigned servers.
44+
*
45+
* @since 1.0
46+
*/
47+
@PublicEvolving
48+
public final class TabletServerDescription {
49+
50+
private final int serverId;
51+
private final int numReplicas;
52+
private final int inSyncReplicas;
53+
private final int numLeaderReplicas;
54+
private final int activeLeaderReplicas;
55+
56+
public TabletServerDescription(
57+
int serverId,
58+
int numReplicas,
59+
int inSyncReplicas,
60+
int numLeaderReplicas,
61+
int activeLeaderReplicas) {
62+
this.serverId = serverId;
63+
this.numReplicas = numReplicas;
64+
this.inSyncReplicas = inSyncReplicas;
65+
this.numLeaderReplicas = numLeaderReplicas;
66+
this.activeLeaderReplicas = activeLeaderReplicas;
67+
}
68+
69+
public int getServerId() {
70+
return serverId;
71+
}
72+
73+
/** Number of replicas assigned to this tablet server. */
74+
public int getNumReplicas() {
75+
return numReplicas;
76+
}
77+
78+
/** Number of assigned replicas that are in the ISR of their bucket. */
79+
public int getInSyncReplicas() {
80+
return inSyncReplicas;
81+
}
82+
83+
/** Number of hosted replicas that are the elected leader of their bucket. */
84+
public int getNumLeaderReplicas() {
85+
return numLeaderReplicas;
86+
}
87+
88+
/** Number of leader replicas on this server that are currently active. */
89+
public int getActiveLeaderReplicas() {
90+
return activeLeaderReplicas;
91+
}
92+
93+
@Override
94+
public boolean equals(Object o) {
95+
if (this == o) {
96+
return true;
97+
}
98+
if (!(o instanceof TabletServerDescription)) {
99+
return false;
100+
}
101+
TabletServerDescription that = (TabletServerDescription) o;
102+
return serverId == that.serverId
103+
&& numReplicas == that.numReplicas
104+
&& inSyncReplicas == that.inSyncReplicas
105+
&& numLeaderReplicas == that.numLeaderReplicas
106+
&& activeLeaderReplicas == that.activeLeaderReplicas;
107+
}
108+
109+
@Override
110+
public int hashCode() {
111+
return Objects.hash(
112+
serverId, numReplicas, inSyncReplicas, numLeaderReplicas, activeLeaderReplicas);
113+
}
114+
115+
@Override
116+
public String toString() {
117+
return "TabletServerDescription{"
118+
+ "serverId="
119+
+ serverId
120+
+ ", numReplicas="
121+
+ numReplicas
122+
+ ", inSyncReplicas="
123+
+ inSyncReplicas
124+
+ ", numLeaderReplicas="
125+
+ numLeaderReplicas
126+
+ ", activeLeaderReplicas="
127+
+ activeLeaderReplicas
128+
+ '}';
129+
}
130+
}

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import org.apache.fluss.client.admin.ClusterHealthStatus;
2222
import org.apache.fluss.client.admin.OffsetSpec;
2323
import org.apache.fluss.client.admin.ProducerOffsetsResult;
24+
import org.apache.fluss.client.admin.TabletServerDescription;
2425
import org.apache.fluss.client.lookup.LookupBatch;
2526
import org.apache.fluss.client.lookup.PrefixLookupBatch;
2627
import org.apache.fluss.client.metadata.AcquireKvSnapshotLeaseResult;
@@ -55,6 +56,7 @@
5556
import org.apache.fluss.rpc.messages.AlterDatabaseRequest;
5657
import org.apache.fluss.rpc.messages.AlterTableRequest;
5758
import org.apache.fluss.rpc.messages.CreatePartitionRequest;
59+
import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
5860
import org.apache.fluss.rpc.messages.DropPartitionRequest;
5961
import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
6062
import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse;
@@ -898,4 +900,18 @@ private static ClusterHealthStatus toClusterHealthStatus(int pbStatus) {
898900
return ClusterHealthStatus.UNKNOWN;
899901
}
900902
}
903+
904+
public static List<TabletServerDescription> toTabletServerDescriptions(
905+
DescribeTabletServersResponse resp) {
906+
return resp.getTabletServersList().stream()
907+
.map(
908+
load ->
909+
new TabletServerDescription(
910+
load.getServerId(),
911+
load.getNumReplicas(),
912+
load.getInSyncReplicas(),
913+
load.getNumLeaderReplicas(),
914+
load.getActiveLeaderReplicas()))
915+
.collect(Collectors.toList());
916+
}
901917
}

‎fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2860,4 +2860,85 @@ void testClusterHealthDuringRollingUpgrade() throws Exception {
28602860
assertThat(afterRecovery.getNumLeaderReplicas())
28612861
.isEqualTo(afterRecovery.getActiveLeaderReplicas());
28622862
}
2863+
2864+
@Test
2865+
void testDescribeTabletServersDuringRollingUpgrade() throws Exception {
2866+
TablePath tablePath = TablePath.of("test_db", "describe_tablet_servers_table");
2867+
TableDescriptor tableDescriptor =
2868+
TableDescriptor.builder().schema(DEFAULT_SCHEMA).distributedBy(3, "id").build();
2869+
long tableId = createTable(tablePath, tableDescriptor, true);
2870+
waitAllReplicasReady(tableId, 3);
2871+
2872+
// Phase 1: Cluster is healthy - every live server is reported, hosts replicas of the
2873+
// created table (replication factor 3 on 3 servers) and is green.
2874+
List<TabletServerDescription> servers = admin.describeTabletServers().get();
2875+
assertThat(servers).extracting(TabletServerDescription::getServerId).contains(0, 1, 2);
2876+
for (TabletServerDescription server : servers) {
2877+
assertThat(server.getNumReplicas()).isGreaterThan(0);
2878+
assertThat(isServerGreen(server)).isTrue();
2879+
}
2880+
2881+
// The per-server counters must sum up to the cluster-wide health counters.
2882+
ClusterHealth health = admin.getClusterHealth().get();
2883+
assertThat(servers.stream().mapToInt(TabletServerDescription::getNumReplicas).sum())
2884+
.isEqualTo(health.getNumReplicas());
2885+
assertThat(servers.stream().mapToInt(TabletServerDescription::getInSyncReplicas).sum())
2886+
.isEqualTo(health.getInSyncReplicas());
2887+
assertThat(servers.stream().mapToInt(TabletServerDescription::getNumLeaderReplicas).sum())
2888+
.isEqualTo(health.getNumLeaderReplicas());
2889+
assertThat(
2890+
servers.stream()
2891+
.mapToInt(TabletServerDescription::getActiveLeaderReplicas)
2892+
.sum())
2893+
.isEqualTo(health.getActiveLeaderReplicas());
2894+
2895+
// Phase 2: Stop one tablet server (simulate server crash during rolling upgrade). It must
2896+
// still be reported with its assigned replicas, but no longer green - an operator must
2897+
// not treat it as safe to remove.
2898+
int stoppedServerId = 0;
2899+
FLUSS_CLUSTER_EXTENSION.stopTabletServer(stoppedServerId);
2900+
FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(2);
2901+
2902+
for (int bucket = 0; bucket < 3; bucket++) {
2903+
TableBucket tb = new TableBucket(tableId, bucket);
2904+
FLUSS_CLUSTER_EXTENSION.waitUntilReplicaShrinkFromIsr(tb, stoppedServerId);
2905+
}
2906+
2907+
TabletServerDescription stopped =
2908+
getTabletServerDescription(admin.describeTabletServers().get(), stoppedServerId);
2909+
assertThat(stopped.getNumReplicas()).isGreaterThan(0);
2910+
assertThat(stopped.getInSyncReplicas()).isLessThan(stopped.getNumReplicas());
2911+
2912+
// Phase 3: Restart the server and wait until every server is green again.
2913+
FLUSS_CLUSTER_EXTENSION.startTabletServer(stoppedServerId);
2914+
FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(3);
2915+
2916+
for (int bucket = 0; bucket < 3; bucket++) {
2917+
TableBucket tb = new TableBucket(tableId, bucket);
2918+
FLUSS_CLUSTER_EXTENSION.waitUntilReplicaExpandToIsr(tb, stoppedServerId);
2919+
}
2920+
2921+
waitUntil(
2922+
() ->
2923+
admin.describeTabletServers().get().stream()
2924+
.allMatch(FlussAdminITCase::isServerGreen),
2925+
Duration.ofMinutes(1),
2926+
"All tablet servers should become green again after server restart");
2927+
}
2928+
2929+
private static TabletServerDescription getTabletServerDescription(
2930+
List<TabletServerDescription> servers, int serverId) {
2931+
return servers.stream()
2932+
.filter(server -> server.getServerId() == serverId)
2933+
.findFirst()
2934+
.orElseThrow(
2935+
() ->
2936+
new AssertionError(
2937+
"no description reported for tablet server " + serverId));
2938+
}
2939+
2940+
private static boolean isServerGreen(TabletServerDescription server) {
2941+
return server.getInSyncReplicas() == server.getNumReplicas()
2942+
&& server.getActiveLeaderReplicas() == server.getNumLeaderReplicas();
2943+
}
28632944
}

‎fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/testutils/TestAdminAdapter.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.apache.fluss.client.admin.OffsetSpec;
2727
import org.apache.fluss.client.admin.ProducerOffsetsResult;
2828
import org.apache.fluss.client.admin.RegisterResult;
29+
import org.apache.fluss.client.admin.TabletServerDescription;
2930
import org.apache.fluss.client.metadata.ActiveKvSnapshots;
3031
import org.apache.fluss.client.metadata.KvSnapshotMetadata;
3132
import org.apache.fluss.client.metadata.KvSnapshots;
@@ -326,6 +327,11 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
326327
throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
327328
}
328329

330+
@Override
331+
public CompletableFuture<List<TabletServerDescription>> describeTabletServers() {
332+
throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
333+
}
334+
329335
@Override
330336
public CompletableFuture<List<RemoteLogManifestInfo>> listRemoteLogManifests(
331337
long tableId, @Nullable Long partitionId) {

‎fluss-rpc/src/main/java/org/apache/fluss/rpc/gateway/AdminReadOnlyGateway.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
2323
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
2424
import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
25+
import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
26+
import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
2527
import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
2628
import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
2729
import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
@@ -202,4 +204,13 @@ CompletableFuture<DescribeClusterConfigsResponse> describeClusterConfigs(
202204
*/
203205
@RPC(api = ApiKeys.GET_CLUSTER_HEALTH)
204206
CompletableFuture<GetClusterHealthResponse> getClusterHealth(GetClusterHealthRequest request);
207+
208+
/**
209+
* Describe the replica load of each tablet server in the cluster.
210+
*
211+
* @return the describe tablet servers response.
212+
*/
213+
@RPC(api = ApiKeys.DESCRIBE_TABLET_SERVERS)
214+
CompletableFuture<DescribeTabletServersResponse> describeTabletServers(
215+
DescribeTabletServersRequest request);
205216
}

‎fluss-rpc/src/main/java/org/apache/fluss/rpc/protocol/ApiKeys.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,8 @@ public enum ApiKeys {
106106
SCAN_KV(1061, 0, 0, PUBLIC),
107107
GET_CLUSTER_HEALTH(1062, 0, 0, PUBLIC),
108108
LIST_REMOTE_LOG_MANIFESTS(1063, 0, 0, PUBLIC),
109-
LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC);
109+
LIST_KV_SNAPSHOTS(1064, 0, 0, PUBLIC),
110+
DESCRIBE_TABLET_SERVERS(1065, 0, 0, PUBLIC);
110111

111112
private static final Map<Integer, ApiKeys> ID_TO_TYPE =
112113
Arrays.stream(ApiKeys.values())

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

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -807,8 +807,23 @@ message GetClusterHealthResponse {
807807
required int32 status = 5; // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
808808
}
809809

810+
message DescribeTabletServersRequest {
811+
}
812+
813+
message DescribeTabletServersResponse {
814+
repeated PbTabletServerLoad tablet_servers = 1;
815+
}
816+
810817

811818
// --------------- Inner classes ----------------
819+
message PbTabletServerLoad {
820+
required int32 server_id = 1;
821+
required int32 num_replicas = 2;
822+
required int32 in_sync_replicas = 3;
823+
required int32 num_leader_replicas = 4;
824+
required int32 active_leader_replicas = 5;
825+
}
826+
812827
message PbApiVersion {
813828
required int32 api_key = 1;
814829
required int32 min_version = 2;

‎fluss-rpc/src/test/java/org/apache/fluss/rpc/TestingTabletGatewayService.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@
2323
import org.apache.fluss.rpc.messages.DatabaseExistsResponse;
2424
import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
2525
import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
26+
import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
27+
import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
2628
import org.apache.fluss.rpc.messages.FetchLogRequest;
2729
import org.apache.fluss.rpc.messages.FetchLogResponse;
2830
import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
@@ -272,4 +274,10 @@ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
272274
GetClusterHealthRequest request) {
273275
return null;
274276
}
277+
278+
@Override
279+
public CompletableFuture<DescribeTabletServersResponse> describeTabletServers(
280+
DescribeTabletServersRequest request) {
281+
return null;
282+
}
275283
}

0 commit comments

Comments
 (0)