From 939c0166a3701605d857cc3fd53a043568399154 Mon Sep 17 00:00:00 2001 From: yunhong <337361684@qq.com> Date: Tue, 27 Oct 2026 09:36:19 +0800 Subject: [PATCH 1/2] [client] Unify cluster inspection in DescribeCluster Unify cluster nodes and tablet server replica loads in Admin.describeCluster, using RPC API key 1075. Keep getServerNodes on the Metadata discovery path and enforce cluster DESCRIBE authorization for cluster inspection. Document coordinator routing and request volume guidance, and update the protocol bindings and coverage for authorization, missing loads and retries. --- .../org/apache/fluss/client/admin/Admin.java | 26 ++- .../client/admin/ClusterDescription.java | 121 ++++++++++ .../apache/fluss/client/admin/FlussAdmin.java | 20 +- .../client/admin/TabletServerDescription.java | 2 +- .../client/utils/ClientRpcMessageUtils.java | 71 ++++-- .../fluss/client/utils/MetadataUtils.java | 124 ----------- .../client/admin/ClusterDescriptionTest.java | 140 ++++++++++++ .../fluss/client/admin/FlussAdmin2ITCase.java | 90 ++++---- .../fluss/client/admin/FlussAdminITCase.java | 14 +- .../fluss/client/admin/FlussAdminTest.java | 206 ++++++++++++++++++ .../acl/FlussAuthorizationITCase.java | 34 +++ .../fluss/client/utils/MetadataUtilsTest.java | 136 ------------ .../procedure/ListServerTagProcedure.java | 8 +- .../sink/testutils/TestAdminAdapter.java | 4 +- .../rpc/gateway/AdminReadOnlyGateway.java | 11 - .../apache/fluss/rpc/protocol/ApiKeys.java | 3 +- fluss-rpc/src/main/proto/FlussApi.proto | 10 +- .../rpc/TestingTabletGatewayService.java | 8 - fluss-rust/crates/fluss/src/proto/fluss.rs | 11 +- .../coordinator/CoordinatorService.java | 43 ++-- .../fluss/server/tablet/TabletService.java | 25 --- ...ersTest.java => TabletServerLoadTest.java} | 34 ++- .../coordinator/TestCoordinatorGateway.java | 8 - .../tablet/TestTabletServerGateway.java | 8 - website/docs/apis/java/index.md | 25 +++ website/docs/engine-flink/procedures.md | 3 +- website/docs/security/authorization.md | 3 +- 27 files changed, 721 insertions(+), 467 deletions(-) create mode 100644 fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterDescription.java create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/admin/ClusterDescriptionTest.java create mode 100644 fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java delete mode 100644 fluss-client/src/test/java/org/apache/fluss/client/utils/MetadataUtilsTest.java rename fluss-server/src/test/java/org/apache/fluss/server/coordinator/{DescribeTabletServersTest.java => TabletServerLoadTest.java} (89%) 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 63c15187013..60ff2fd49fe 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 @@ -55,6 +55,7 @@ import org.apache.fluss.exception.TableNotPartitionedException; import org.apache.fluss.exception.TooManyBucketsException; import org.apache.fluss.exception.TooManyPartitionsException; +import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.metadata.BucketInfo; import org.apache.fluss.metadata.DatabaseChange; import org.apache.fluss.metadata.DatabaseDescriptor; @@ -93,10 +94,6 @@ public interface Admin extends AutoCloseable { /** * Get the current server node information asynchronously: the coordinator server, followed by * the alive tablet servers. - * - *

Each tablet server carries its {@link ServerTag} in {@link ServerNode#serverTag()}, or - * {@code null} if it has none. The tags are read from the coordinator. Against a cluster older - * than Fluss 1.1, which cannot report server tags, the server nodes are returned without tags. */ CompletableFuture> getServerNodes(); @@ -929,10 +926,21 @@ CompletableFuture registerProducerOffsets( CompletableFuture getClusterHealth(); /** - * Describe the replica load of each tablet server in the cluster asynchronously. + * Describe the coordinator, live tablet servers and per-server replica loads asynchronously. + * + *

The returned {@link ClusterDescription} provides node addresses, racks and server tags, + * together with replica loads keyed by server ID. Live servers have explicit loads even when + * they host no replicas. Loads also include offline servers with assigned replicas. A missing + * load means statistics are unavailable and must never be interpreted as zero replicas. + * + *

This API requires DESCRIBE permission on the cluster resource. Retriable failures trigger + * a metadata refresh and one retry to handle coordinator address changes. Servers that do not + * support this API fail with {@link UnsupportedVersionException}. * - *

The returned list contains {@link TabletServerDescription} counters per tablet server. - * Live tablet servers are always included, even when they host no replicas. + *

Requests currently go directly to the coordinator, which aggregates replica loads on its + * event thread. Avoid high request volumes or frequent polling, as these can delay other + * coordinator work. A future implementation may route requests to a randomly selected tablet + * server to reduce coordinator load. * *

This API is designed for operational tooling (e.g., a Kubernetes operator): * @@ -946,10 +954,10 @@ CompletableFuture registerProducerOffsets( *

  • reporting per-server tablet load in cluster status. * * - * @return a {@link CompletableFuture} that completes with the per tablet server replica loads. + * @return a {@link CompletableFuture} that completes with the cluster description. * @since 1.1 */ - CompletableFuture> describeTabletServers(); + CompletableFuture describeCluster(); /** * List per-bucket remote log manifest entries for a table or partition scope. diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterDescription.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterDescription.java new file mode 100644 index 00000000000..d522a5c4849 --- /dev/null +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/ClusterDescription.java @@ -0,0 +1,121 @@ +/* + * 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 org.apache.fluss.cluster.ServerNode; + +import javax.annotation.Nullable; + +import java.util.Collections; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.TreeMap; + +/** + * Cluster information returned by {@link Admin#describeCluster()}. + * + *

    Both tablet server maps are keyed by server ID. The node map contains only live servers, + * including their addresses, racks and tags. The load map also contains offline servers that still + * have assigned replicas. A missing load entry means statistics are unavailable, never that the + * server has zero replicas or is safe to remove. + * + * @since 1.1 + */ +@PublicEvolving +public final class ClusterDescription { + + @Nullable private final ServerNode coordinatorServer; + private final Map tabletServers; + private final Map tabletServerLoads; + + /** Creates a cluster description with defensive copies of the server maps. */ + public ClusterDescription( + @Nullable ServerNode coordinatorServer, + Map tabletServers, + Map tabletServerLoads) { + this.coordinatorServer = coordinatorServer; + this.tabletServers = Collections.unmodifiableMap(new TreeMap<>(tabletServers)); + this.tabletServerLoads = Collections.unmodifiableMap(new TreeMap<>(tabletServerLoads)); + } + + /** Returns the coordinator, or an empty optional if it was not reported. */ + public Optional getCoordinatorServer() { + return Optional.ofNullable(coordinatorServer); + } + + /** Returns the live tablet servers by server ID, in an immutable map ordered by ID. */ + public Map getTabletServers() { + return tabletServers; + } + + /** + * Returns the reported replica loads by server ID, in an immutable map ordered by ID. + * + *

    Live servers with no replicas have an explicit zero load. Offline servers with assigned + * replicas are also included. Missing entries indicate unavailable statistics; in particular, a + * server that does not report replica loads may return an empty map. + */ + public Map getTabletServerLoads() { + return tabletServerLoads; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof ClusterDescription)) { + return false; + } + ClusterDescription that = (ClusterDescription) o; + return Objects.equals(coordinatorServer, that.coordinatorServer) + && tabletServers.equals(that.tabletServers) + // ServerNode equality describes node identity and does not compare server tags. + && tabletServers.entrySet().stream() + .allMatch( + entry -> + entry.getValue().serverTag() + == that.tabletServers + .get(entry.getKey()) + .serverTag()) + && tabletServerLoads.equals(that.tabletServerLoads); + } + + @Override + public int hashCode() { + int tagsHash = 1; + for (ServerNode node : tabletServers.values()) { + tagsHash = 31 * tagsHash + Objects.hashCode(node.serverTag()); + } + return Objects.hash(coordinatorServer, tabletServers, tabletServerLoads, tagsHash); + } + + @Override + public String toString() { + return "ClusterDescription{" + + "coordinatorServer=" + + coordinatorServer + + ", tabletServers=" + + tabletServers + + ", tabletServerLoads=" + + tabletServerLoads + + '}'; + } +} 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 1c991d10e2c..902c5788a3c 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 @@ -75,7 +75,7 @@ import org.apache.fluss.rpc.messages.DeleteProducerOffsetsRequest; import org.apache.fluss.rpc.messages.DescribeBucketsRequest; import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest; -import org.apache.fluss.rpc.messages.DescribeTabletServersRequest; +import org.apache.fluss.rpc.messages.DescribeClusterRequest; import org.apache.fluss.rpc.messages.DropAclsRequest; import org.apache.fluss.rpc.messages.DropDatabaseRequest; import org.apache.fluss.rpc.messages.DropTableRequest; @@ -144,8 +144,6 @@ import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makePbPartitionSpec; import static org.apache.fluss.client.utils.ClientRpcMessageUtils.makeRegisterProducerOffsetsRequest; import static org.apache.fluss.client.utils.ClientRpcMessageUtils.toConfigEntries; -import static org.apache.fluss.client.utils.MetadataUtils.fallbackIfUnsupported; -import static org.apache.fluss.client.utils.MetadataUtils.sendDescribeClusterRequest; import static org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster; import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toAclBindings; import static org.apache.fluss.rpc.util.CommonRpcMessageUtils.toPbAclBindingFilters; @@ -212,15 +210,6 @@ public FlussAdmin(RpcClient client, MetadataUpdater metadataUpdater) { @Override public CompletableFuture> getServerNodes() { - // Only the coordinator knows the server tags. A coordinator older than Fluss 1.1 does not - // support the describe cluster request, so fall back to the metadata request, which returns - // the same server nodes without tags. - return fallbackIfUnsupported( - () -> sendDescribeClusterRequest(retryableCoordinatorGateway), - this::getServerNodesFromMetadata); - } - - private CompletableFuture> getServerNodesFromMetadata() { CompletableFuture> future = new CompletableFuture<>(); CompletableFuture.runAsync( () -> { @@ -1211,9 +1200,10 @@ public CompletableFuture getClusterHealth() { } @Override - public CompletableFuture> describeTabletServers() { - return gateway.describeTabletServers(new DescribeTabletServersRequest()) - .thenApply(ClientRpcMessageUtils::toTabletServerDescriptions); + public CompletableFuture describeCluster() { + return retryableCoordinatorGateway + .describeCluster(new DescribeClusterRequest()) + .thenApply(ClientRpcMessageUtils::toClusterDescription); } @VisibleForTesting diff --git a/fluss-client/src/main/java/org/apache/fluss/client/admin/TabletServerDescription.java b/fluss-client/src/main/java/org/apache/fluss/client/admin/TabletServerDescription.java index a195985d3a8..90d1dda22a4 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/admin/TabletServerDescription.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/admin/TabletServerDescription.java @@ -22,7 +22,7 @@ import java.util.Objects; /** - * Per tablet server replica load returned by {@link Admin#describeTabletServers()}. + * Per tablet server replica load returned by {@link Admin#describeCluster()}. * *

    It reports the replica counters of a single tablet server: how many replicas it hosts, how * many of those are in sync, how many buckets it leads, and how many of those leaders are active. 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 8f598b62dfd..0147bfef1bf 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,7 @@ package org.apache.fluss.client.utils; +import org.apache.fluss.client.admin.ClusterDescription; import org.apache.fluss.client.admin.ClusterHealth; import org.apache.fluss.client.admin.ClusterHealthStatus; import org.apache.fluss.client.admin.OffsetSpec; @@ -33,10 +34,13 @@ import org.apache.fluss.client.write.KvWriteBatch; import org.apache.fluss.client.write.ReadyWriteBatch; import org.apache.fluss.cluster.Cluster; +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.cluster.ServerType; import org.apache.fluss.cluster.rebalance.RebalancePlanForBucket; import org.apache.fluss.cluster.rebalance.RebalanceProgress; import org.apache.fluss.cluster.rebalance.RebalanceResultForBucket; import org.apache.fluss.cluster.rebalance.RebalanceStatus; +import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.config.cluster.AlterConfigOpType; import org.apache.fluss.config.cluster.ColumnPositionType; import org.apache.fluss.config.cluster.ConfigEntry; @@ -60,7 +64,7 @@ import org.apache.fluss.rpc.messages.AlterTableRequest; import org.apache.fluss.rpc.messages.CreatePartitionRequest; import org.apache.fluss.rpc.messages.DescribeBucketsResponse; -import org.apache.fluss.rpc.messages.DescribeTabletServersResponse; +import org.apache.fluss.rpc.messages.DescribeClusterResponse; import org.apache.fluss.rpc.messages.DropPartitionRequest; import org.apache.fluss.rpc.messages.GetClusterHealthResponse; import org.apache.fluss.rpc.messages.GetFileSystemSecurityTokenResponse; @@ -103,8 +107,10 @@ import org.apache.fluss.rpc.messages.PbRemoteLogManifestEntry; import org.apache.fluss.rpc.messages.PbRemotePathAndLocalFile; import org.apache.fluss.rpc.messages.PbRenameColumn; +import org.apache.fluss.rpc.messages.PbServerNode; import org.apache.fluss.rpc.messages.PbTableBucket; import org.apache.fluss.rpc.messages.PbTableStatsReqForBucket; +import org.apache.fluss.rpc.messages.PbTabletServerLoad; import org.apache.fluss.rpc.messages.PrefixLookupRequest; import org.apache.fluss.rpc.messages.ProduceLogRequest; import org.apache.fluss.rpc.messages.PutKvRequest; @@ -114,6 +120,9 @@ import org.apache.fluss.utils.json.DataTypeJsonSerde; import org.apache.fluss.utils.json.JsonSerdeUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import javax.annotation.Nullable; import java.util.ArrayList; @@ -138,6 +147,8 @@ */ public class ClientRpcMessageUtils { + private static final Logger LOG = LoggerFactory.getLogger(ClientRpcMessageUtils.class); + public static ProduceLogRequest makeProduceLogRequest( long tableId, int acks, int maxRequestTimeoutMs, List readyBatches) { ProduceLogRequest request = @@ -987,18 +998,50 @@ private static ClusterHealthStatus toClusterHealthStatus(int pbStatus) { } } - public static List toTabletServerDescriptions( - DescribeTabletServersResponse resp) { - return resp.getTabletServersList().stream() - .map( - load -> - new TabletServerDescription( - load.getServerId(), - load.getNumReplicas(), - load.getInSyncReplicas(), - load.getNumLeaderReplicas(), - load.getActiveLeaderReplicas(), - load.getLeaderlessReplicas())) - .collect(Collectors.toList()); + /** Converts cluster information without inferring zero loads for missing entries. */ + public static ClusterDescription toClusterDescription(DescribeClusterResponse response) { + ServerNode coordinator = + response.hasCoordinatorServer() + ? toServerNode(response.getCoordinatorServer(), ServerType.COORDINATOR) + : null; + Map tabletServers = new HashMap<>(); + for (PbServerNode node : response.getTabletServersList()) { + tabletServers.put(node.getNodeId(), toServerNode(node, ServerType.TABLET_SERVER)); + } + Map loads = new HashMap<>(); + for (PbTabletServerLoad load : response.getTabletServerLoadsList()) { + loads.put( + load.getServerId(), + new TabletServerDescription( + load.getServerId(), + load.getNumReplicas(), + load.getInSyncReplicas(), + load.getNumLeaderReplicas(), + load.getActiveLeaderReplicas(), + load.getLeaderlessReplicas())); + } + return new ClusterDescription(coordinator, tabletServers, loads); + } + + private static ServerNode toServerNode(PbServerNode node, ServerType type) { + return new ServerNode( + node.getNodeId(), + node.getHost(), + node.getPort(), + type, + node.hasRack() ? node.getRack() : null, + node.hasServerTag() ? toServerTag(node.getServerTag()) : null); + } + + @Nullable + private static ServerTag toServerTag(int value) { + try { + return ServerTag.valueOf(value); + } catch (IllegalArgumentException e) { + LOG.warn( + "Ignoring unknown server tag {}, it was probably added by a newer Fluss version.", + value); + return null; + } } } diff --git a/fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java b/fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java index 59087660a50..c56dd1169b3 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java @@ -17,24 +17,18 @@ package org.apache.fluss.client.utils; -import org.apache.fluss.annotation.VisibleForTesting; import org.apache.fluss.cluster.BucketLocation; import org.apache.fluss.cluster.Cluster; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; -import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.exception.StaleMetadataException; -import org.apache.fluss.exception.UnsupportedVersionException; import org.apache.fluss.metadata.PhysicalTablePath; import org.apache.fluss.metadata.TableBucket; import org.apache.fluss.metadata.TableOrPartition; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.rpc.GatewayClientProxy; import org.apache.fluss.rpc.RpcClient; -import org.apache.fluss.rpc.gateway.AdminGateway; import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway; -import org.apache.fluss.rpc.messages.DescribeClusterRequest; -import org.apache.fluss.rpc.messages.DescribeClusterResponse; import org.apache.fluss.rpc.messages.MetadataRequest; import org.apache.fluss.rpc.messages.MetadataResponse; import org.apache.fluss.rpc.messages.PbBucketMetadata; @@ -42,10 +36,6 @@ import org.apache.fluss.rpc.messages.PbServerNode; import org.apache.fluss.rpc.messages.PbTableMetadata; import org.apache.fluss.rpc.messages.PbTablePath; -import org.apache.fluss.utils.ExceptionUtils; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nullable; @@ -56,17 +46,13 @@ import java.util.Map; import java.util.Random; import java.util.Set; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; -import java.util.function.Supplier; /** Utils for metadata for client. */ public class MetadataUtils { - private static final Logger LOG = LoggerFactory.getLogger(MetadataUtils.class); - private static final Random randOffset = new Random(); /** @@ -178,102 +164,6 @@ public static Cluster sendMetadataRequestAndRebuildCluster( // time out here } - /** - * Sends a {@link DescribeClusterRequest} to the coordinator. The returned server nodes are the - * coordinator server, followed by the alive tablet servers with their {@link ServerTag}. - */ - public static CompletableFuture> sendDescribeClusterRequest( - AdminGateway gateway) { - return gateway.describeCluster(new DescribeClusterRequest()) - .thenApply(MetadataUtils::toServerNodes); - } - - /** - * Runs {@code request}, or {@code fallback} instead when the server does not support the - * request, for example because it runs an older Fluss version. Any other failure of {@code - * request} fails the returned future. - */ - public static CompletableFuture fallbackIfUnsupported( - Supplier> request, Supplier> fallback) { - CompletableFuture result = new CompletableFuture<>(); - CompletableFuture requestFuture; - try { - requestFuture = request.get(); - } catch (Throwable t) { - requestFuture = new CompletableFuture<>(); - requestFuture.completeExceptionally(t); - } - requestFuture.whenComplete( - (value, error) -> { - if (error == null) { - result.complete(value); - } else if (!ExceptionUtils.findThrowable( - error, UnsupportedVersionException.class) - .isPresent()) { - result.completeExceptionally( - ExceptionUtils.stripCompletionException(error)); - } else { - LOG.info( - "The server does not support the request, falling back: {}", - error.getMessage()); - try { - fallback.get() - .whenComplete( - (fallbackValue, fallbackError) -> { - if (fallbackError == null) { - result.complete(fallbackValue); - } else { - result.completeExceptionally( - ExceptionUtils.stripCompletionException( - fallbackError)); - } - }); - } catch (Throwable t) { - result.completeExceptionally(t); - } - } - }); - return result; - } - - @VisibleForTesting - static List toServerNodes(DescribeClusterResponse response) { - List serverNodes = new ArrayList<>(); - ServerNode coordinator = getCoordinatorServer(response); - if (coordinator != null) { - serverNodes.add(coordinator); - } - for (PbServerNode serverNode : response.getTabletServersList()) { - serverNodes.add( - new ServerNode( - serverNode.getNodeId(), - serverNode.getHost(), - serverNode.getPort(), - ServerType.TABLET_SERVER, - serverNode.hasRack() ? serverNode.getRack() : null, - serverNode.hasServerTag() - ? toServerTag(serverNode.getServerTag()) - : null)); - } - return serverNodes; - } - - /** - * Converts the server tag value of a response. A value this client does not know, written by a - * newer server, is reported as no tag instead of failing the whole request. - */ - @Nullable - private static ServerTag toServerTag(int value) { - try { - return ServerTag.valueOf(value); - } catch (IllegalArgumentException e) { - LOG.warn( - "Ignoring unknown server tag {}, it was probably added by a newer Fluss version.", - value); - return null; - } - } - private static NewTableMetadata getTableMetadataToUpdate( Cluster cluster, MetadataResponse metadataResponse) { Map newTablePathToTableId = new HashMap<>(); @@ -414,20 +304,6 @@ private static ServerNode getCoordinatorServer(MetadataResponse response) { } } - @Nullable - private static ServerNode getCoordinatorServer(DescribeClusterResponse response) { - if (!response.hasCoordinatorServer()) { - return null; - } else { - PbServerNode protoServerNode = response.getCoordinatorServer(); - return new ServerNode( - protoServerNode.getNodeId(), - protoServerNode.getHost(), - protoServerNode.getPort(), - ServerType.COORDINATOR); - } - } - private static Map getAliveTabletServers(MetadataResponse response) { Map aliveTabletServers = new HashMap<>(); response.getTabletServersList() diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/ClusterDescriptionTest.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/ClusterDescriptionTest.java new file mode 100644 index 00000000000..03b62deb7ea --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/ClusterDescriptionTest.java @@ -0,0 +1,140 @@ +/* + * 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.client.utils.ClientRpcMessageUtils; +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.cluster.rebalance.ServerTag; +import org.apache.fluss.rpc.messages.DescribeClusterResponse; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link ClusterDescription} and its RPC conversion. */ +class ClusterDescriptionTest { + + @Test + void testLiveNodesAndOfflineLoads() { + DescribeClusterResponse response = new DescribeClusterResponse(); + response.setCoordinatorServer().setNodeId(0).setHost("coordinator").setPort(9123); + response.addTabletServer() + .setNodeId(2) + .setHost("ts-2") + .setPort(9125) + .setRack("rack-2") + .setServerTag(ServerTag.PERMANENT_OFFLINE.value); + response.addTabletServer().setNodeId(0).setHost("ts-0").setPort(9124); + response.addTabletServer().setNodeId(3).setHost("ts-3").setPort(9126).setServerTag(42); + response.addTabletServerLoad() + .setServerId(5) + .setNumReplicas(3) + .setInSyncReplicas(1) + .setNumLeaderReplicas(2) + .setActiveLeaderReplicas(1) + .setLeaderlessReplicas(1); + response.addTabletServerLoad() + .setServerId(2) + .setNumReplicas(0) + .setInSyncReplicas(0) + .setNumLeaderReplicas(0) + .setActiveLeaderReplicas(0) + .setLeaderlessReplicas(0); + + ClusterDescription description = ClientRpcMessageUtils.toClusterDescription(response); + + assertThat(description.getCoordinatorServer()) + .contains(new ServerNode(0, "coordinator", 9123, ServerType.COORDINATOR)); + assertThat(description.getTabletServers().keySet()).containsExactly(0, 2, 3); + ServerNode taggedNode = description.getTabletServers().get(2); + assertThat(taggedNode.host()).isEqualTo("ts-2"); + assertThat(taggedNode.port()).isEqualTo(9125); + assertThat(taggedNode.rack()).isEqualTo("rack-2"); + assertThat(taggedNode.serverTag()).isEqualTo(ServerTag.PERMANENT_OFFLINE); + assertThat(description.getTabletServers().get(0).serverTag()).isNull(); + assertThat(description.getTabletServers().get(3).serverTag()).isNull(); + assertThat(description.getTabletServerLoads().keySet()).containsExactly(2, 5); + assertThat(description.getTabletServerLoads().get(2)) + .isEqualTo(new TabletServerDescription(2, 0, 0, 0, 0, 0)); + assertThat(description.getTabletServerLoads().get(5)) + .isEqualTo(new TabletServerDescription(5, 3, 1, 2, 1, 1)); + assertThat(description.getTabletServerLoads()).doesNotContainKey(0); + } + + @Test + void testMissingStatisticsAreNotZeroLoads() { + DescribeClusterResponse response = new DescribeClusterResponse(); + response.addTabletServer().setNodeId(1).setHost("ts-1").setPort(9124); + + ClusterDescription description = ClientRpcMessageUtils.toClusterDescription(response); + + assertThat(description.getCoordinatorServer()).isEmpty(); + assertThat(description.getTabletServers()).containsOnlyKeys(1); + assertThat(description.getTabletServerLoads()).isEmpty(); + assertThat(description.getTabletServerLoads().get(1)).isNull(); + } + + @Test + void testImmutableSnapshot() { + Map nodes = new HashMap<>(); + nodes.put(1, new ServerNode(1, "ts-1", 9124, ServerType.TABLET_SERVER)); + Map loads = new HashMap<>(); + loads.put(1, new TabletServerDescription(1, 0, 0, 0, 0, 0)); + ClusterDescription description = new ClusterDescription(null, nodes, loads); + ClusterDescription sameDescription = new ClusterDescription(null, nodes, loads); + + nodes.clear(); + loads.clear(); + + assertThat(description.getTabletServers()).hasSize(1); + assertThat(description.getTabletServerLoads()).hasSize(1); + assertThatThrownBy(() -> description.getTabletServers().clear()) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> description.getTabletServerLoads().clear()) + .isInstanceOf(UnsupportedOperationException.class); + assertThat(description).isEqualTo(sameDescription).hasSameHashCodeAs(sameDescription); + } + + @Test + void testEqualityIncludesServerTags() { + ServerNode node = new ServerNode(1, "ts-1", 9124, ServerType.TABLET_SERVER); + ServerNode taggedNode = + new ServerNode( + 1, + "ts-1", + 9124, + ServerType.TABLET_SERVER, + null, + ServerTag.PERMANENT_OFFLINE); + ClusterDescription description = + new ClusterDescription( + null, Collections.singletonMap(1, node), Collections.emptyMap()); + ClusterDescription taggedDescription = + new ClusterDescription( + null, Collections.singletonMap(1, taggedNode), Collections.emptyMap()); + + assertThat(description).isNotEqualTo(taggedDescription); + } +} diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdmin2ITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdmin2ITCase.java index 1f1d0561695..96d95d9b2ad 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdmin2ITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdmin2ITCase.java @@ -19,6 +19,7 @@ package org.apache.fluss.client.admin; +import org.apache.fluss.client.FlussConnection; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.rebalance.ServerTag; import org.apache.fluss.config.AutoPartitionTimeUnit; @@ -35,9 +36,12 @@ import org.apache.fluss.types.DataTypes; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import java.time.Duration; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.Map; @@ -173,8 +177,8 @@ void testListPartitionInfos() throws Exception { } @Test - void testDescribeTabletServersDuringRollingUpgrade() throws Exception { - TablePath tablePath = TablePath.of("test_db", "describe_tablet_servers_table"); + void testDescribeClusterDuringRollingUpgrade() throws Exception { + TablePath tablePath = TablePath.of("test_db", "describe_cluster_table"); long tableId = createTable(tablePath, DATA1_TABLE_DESCRIPTOR_PK, true); waitAllReplicasReady(tableId, 3); @@ -184,12 +188,16 @@ void testDescribeTabletServersDuringRollingUpgrade() throws Exception { // before taking the snapshot asserted below. waitUntil( () -> - admin.describeTabletServers().get().stream() + admin.describeCluster().get().getTabletServerLoads().values().stream() .allMatch(FlussAdmin2ITCase::isServerGreen), Duration.ofMinutes(1), "All tablet servers should be green before the rolling upgrade starts"); - List servers = admin.describeTabletServers().get(); + ClusterDescription description = admin.describeCluster().get(); + assertThat(description.getCoordinatorServer()) + .contains(FLUSS_CLUSTER_EXTENSION.getCoordinatorServerNode()); + assertThat(description.getTabletServers()).containsOnlyKeys(0, 1, 2); + Collection servers = description.getTabletServerLoads().values(); assertThat(servers).extracting(TabletServerDescription::getServerId).contains(0, 1, 2); for (TabletServerDescription server : servers) { assertThat(server.getNumReplicas()).isGreaterThan(0); @@ -201,20 +209,25 @@ void testDescribeTabletServersDuringRollingUpgrade() throws Exception { // not treat it as safe to remove. int stoppedServerId = 0; FLUSS_CLUSTER_EXTENSION.stopTabletServer(stoppedServerId); - FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(2); + try { + FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(2); - for (int bucket = 0; bucket < 3; bucket++) { - TableBucket tb = new TableBucket(tableId, bucket); - FLUSS_CLUSTER_EXTENSION.waitUntilReplicaShrinkFromIsr(tb, stoppedServerId); - } + for (int bucket = 0; bucket < 3; bucket++) { + TableBucket tb = new TableBucket(tableId, bucket); + FLUSS_CLUSTER_EXTENSION.waitUntilReplicaShrinkFromIsr(tb, stoppedServerId); + } - TabletServerDescription stopped = - getTabletServerDescription(admin.describeTabletServers().get(), stoppedServerId); - assertThat(stopped.getNumReplicas()).isGreaterThan(0); - assertThat(stopped.getInSyncReplicas()).isLessThan(stopped.getNumReplicas()); + description = admin.describeCluster().get(); + assertThat(description.getTabletServers()).doesNotContainKey(stoppedServerId); + assertThat(description.getTabletServerLoads()).containsKey(stoppedServerId); + TabletServerDescription stopped = + description.getTabletServerLoads().get(stoppedServerId); + assertThat(stopped.getNumReplicas()).isGreaterThan(0); + assertThat(stopped.getInSyncReplicas()).isLessThan(stopped.getNumReplicas()); + } finally { + FLUSS_CLUSTER_EXTENSION.startTabletServer(stoppedServerId); + } - // Phase 3: Restart the server and wait until every server is green again. - FLUSS_CLUSTER_EXTENSION.startTabletServer(stoppedServerId); FLUSS_CLUSTER_EXTENSION.assertHasTabletServerNumber(3); for (int bucket = 0; bucket < 3; bucket++) { @@ -224,35 +237,49 @@ void testDescribeTabletServersDuringRollingUpgrade() throws Exception { waitUntil( () -> - admin.describeTabletServers().get().stream() + admin.describeCluster().get().getTabletServerLoads().values().stream() .allMatch(FlussAdmin2ITCase::isServerGreen), Duration.ofMinutes(1), "All tablet servers should become green again after server restart"); } - @Test - void testGetServerNodesAfterCoordinatorServerRestart() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testClusterDiscoveryAfterCoordinatorServerRestart(boolean describeCluster) + throws Exception { admin.addServerTag(Collections.singletonList(0), ServerTag.TEMPORARY_OFFLINE).get(); try { ServerNode staleCoordinator = FLUSS_CLUSTER_EXTENSION.getCoordinatorServerNode(); - assertThat(admin.getServerNodes().get()).contains(staleCoordinator); + assertThat(admin.describeCluster().get().getCoordinatorServer()) + .contains(staleCoordinator); - // The same admin keeps the stale coordinator address, so getServerNodes must refresh - // the metadata and retry to reach the restarted coordinator. + // Reuse the same admin with stale coordinator metadata after the address changes. restartCoordinatorServer(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient()); ServerNode newCoordinator = FLUSS_CLUSTER_EXTENSION.getCoordinatorServerNode(); assertThat(newCoordinator).isNotEqualTo(staleCoordinator); - List serverNodes = admin.getServerNodes().get(); + List serverNodes; + if (describeCluster) { + ClusterDescription description = admin.describeCluster().get(); + assertThat(description.getCoordinatorServer()).contains(newCoordinator); + assertThat(description.getTabletServerLoads()).containsOnlyKeys(0, 1, 2); + assertThat(description.getTabletServers().values()) + .filteredOn(node -> node.serverTag() != null) + .extracting(ServerNode::id, ServerNode::serverTag) + .containsExactly(tuple(0, ServerTag.TEMPORARY_OFFLINE)); + serverNodes = new ArrayList<>(description.getTabletServers().values()); + serverNodes.add(description.getCoordinatorServer().get()); + } else { + serverNodes = admin.getServerNodes().get(); + assertThat(serverNodes).allMatch(node -> node.serverTag() == null); + } List expectedNodes = new ArrayList<>(); expectedNodes.add(newCoordinator); expectedNodes.addAll(FLUSS_CLUSTER_EXTENSION.getTabletServerNodes()); assertThat(serverNodes).containsExactlyInAnyOrderElementsOf(expectedNodes); - assertThat(serverNodes) - .filteredOn(node -> node.serverTag() != null) - .extracting(ServerNode::id, ServerNode::serverTag) - .containsExactly(tuple(0, ServerTag.TEMPORARY_OFFLINE)); } finally { + // Discovery can return fresh nodes without updating the cached mutation gateway. + ((FlussConnection) conn).getMetadataUpdater().refreshClusterUntilAvailable(); admin.removeServerTag(Collections.singletonList(0), ServerTag.TEMPORARY_OFFLINE).get(); } } @@ -271,17 +298,6 @@ private static ListPartitionInfosRequest requestFor( return request; } - private static TabletServerDescription getTabletServerDescription( - List servers, int serverId) { - return servers.stream() - .filter(server -> server.getServerId() == serverId) - .findFirst() - .orElseThrow( - () -> - new AssertionError( - "no description reported for tablet server " + serverId)); - } - private static boolean isServerGreen(TabletServerDescription server) { return server.getInSyncReplicas() == server.getNumReplicas() && server.getActiveLeaderReplicas() == server.getNumLeaderReplicas() 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 a978753e43f..d65d6a52812 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 @@ -2331,8 +2331,8 @@ public void testAddAndRemoveServerTags() throws Exception { assertThat(zkClient.getServerTags().get().getServerTags()) .containsEntry(0, ServerTag.PERMANENT_OFFLINE) .containsEntry(1, ServerTag.PERMANENT_OFFLINE); - // getServerNodes reports the same tags, and no tag for server 2 - assertThat(serverTagsOf(admin.getServerNodes().get())) + // describeCluster reports the same tags, and no tag for server 2 + assertThat(serverTagsOf(admin.describeCluster().get().getTabletServers().values())) .containsOnly( entry(0, ServerTag.PERMANENT_OFFLINE), entry(1, ServerTag.PERMANENT_OFFLINE)); @@ -2355,7 +2355,8 @@ public void testAddAndRemoveServerTags() throws Exception { assertThat(serverTags) .containsEntry(0, ServerTag.PERMANENT_OFFLINE) .containsEntry(1, ServerTag.PERMANENT_OFFLINE); - assertThat(serverTagsOf(admin.getServerNodes().get())).isEqualTo(serverTags); + assertThat(serverTagsOf(admin.describeCluster().get().getTabletServers().values())) + .isEqualTo(serverTags); // 4.remove server tag for server 100 admin.removeServerTag(Collections.singletonList(100), ServerTag.PERMANENT_OFFLINE).get(); @@ -2373,7 +2374,8 @@ public void testAddAndRemoveServerTags() throws Exception { admin.removeServerTag(Arrays.asList(0, 1), ServerTag.PERMANENT_OFFLINE).get(); assertThat(zkClient.getServerTags()).isNotPresent(); - assertThat(serverTagsOf(admin.getServerNodes().get())).isEmpty(); + assertThat(serverTagsOf(admin.describeCluster().get().getTabletServers().values())) + .isEmpty(); // restore after test, or else will influence other tests coordinatorContext.addLiveTabletServer(tmpServerInfo); @@ -2392,14 +2394,14 @@ public void testAddAndRemoveServerTags() throws Exception { .hasMessageContaining( "Server tag PERMANENT_OFFLINE not exists for server 2, the current " + "server tag of this server is TEMPORARY_OFFLINE."); - assertThat(serverTagsOf(admin.getServerNodes().get())) + assertThat(serverTagsOf(admin.describeCluster().get().getTabletServers().values())) .containsOnly(entry(2, ServerTag.TEMPORARY_OFFLINE)); // clear server tag admin.removeServerTag(Collections.singletonList(2), ServerTag.TEMPORARY_OFFLINE).get(); } /** Returns the server tags of the tagged server nodes by server id. */ - private static Map serverTagsOf(List serverNodes) { + private static Map serverTagsOf(Collection serverNodes) { return serverNodes.stream() .filter(node -> node.serverTag() != null) .collect(Collectors.toMap(ServerNode::id, ServerNode::serverTag)); diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java new file mode 100644 index 00000000000..6a40279aaea --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java @@ -0,0 +1,206 @@ +/* + * 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.client.metadata.MetadataUpdater; +import org.apache.fluss.cluster.Cluster; +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.cluster.ServerType; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.exception.AuthorizationException; +import org.apache.fluss.exception.NetworkException; +import org.apache.fluss.exception.UnsupportedVersionException; +import org.apache.fluss.rpc.RpcClient; +import org.apache.fluss.rpc.messages.ApiMessage; +import org.apache.fluss.rpc.messages.DescribeClusterResponse; +import org.apache.fluss.rpc.messages.MetadataResponse; +import org.apache.fluss.rpc.protocol.ApiKeys; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiFunction; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for cluster discovery and inspection through {@link FlussAdmin}. */ +class FlussAdminTest { + + private static final ServerNode OLD_COORDINATOR = + new ServerNode(0, "old-coordinator", 9123, ServerType.COORDINATOR); + private static final ServerNode NEW_COORDINATOR = + new ServerNode(0, "new-coordinator", 9123, ServerType.COORDINATOR); + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testDescribeClusterRefreshesAndRetriesOnce(boolean retrySucceeds) throws Exception { + AtomicInteger refreshes = new AtomicInteger(); + List contactedNodes = new ArrayList<>(); + TestingRpcClient client = + new TestingRpcClient( + (node, api) -> { + assertThat(api).isEqualTo(ApiKeys.DESCRIBE_CLUSTER); + contactedNodes.add(node); + if (node.equals(OLD_COORDINATOR) || !retrySucceeds) { + return failed(new NetworkException("coordinator unavailable")); + } + DescribeClusterResponse response = new DescribeClusterResponse(); + response.setCoordinatorServer() + .setNodeId(node.id()) + .setHost(node.host()) + .setPort(node.port()); + return CompletableFuture.completedFuture(response); + }); + MetadataUpdater updater = + new MetadataUpdater(client, new Configuration(), Cluster.empty()) { + @Override + public ServerNode getCoordinatorServer() { + return refreshes.get() == 0 ? OLD_COORDINATOR : NEW_COORDINATOR; + } + + @Override + public void refreshClusterUntilAvailable() { + refreshes.incrementAndGet(); + } + }; + try (FlussAdmin admin = new FlussAdmin(client, updater)) { + if (retrySucceeds) { + assertThat(admin.describeCluster().get(10, TimeUnit.SECONDS).getCoordinatorServer()) + .contains(NEW_COORDINATOR); + } else { + assertThatThrownBy(() -> admin.describeCluster().get(10, TimeUnit.SECONDS)) + .rootCause() + .isInstanceOf(NetworkException.class); + } + assertThat(contactedNodes).containsExactly(OLD_COORDINATOR, NEW_COORDINATOR); + assertThat(refreshes).hasValue(1); + } + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testDescribeClusterDoesNotHideNonRetriableFailures(boolean unsupported) throws Exception { + RuntimeException error = + unsupported + ? new UnsupportedVersionException("DescribeCluster is unavailable") + : new AuthorizationException("cluster DESCRIBE denied"); + AtomicInteger requests = new AtomicInteger(); + TestingRpcClient client = + new TestingRpcClient( + (node, api) -> { + assertThat(api).isEqualTo(ApiKeys.DESCRIBE_CLUSTER); + requests.incrementAndGet(); + return failed(error); + }); + MetadataUpdater updater = + new MetadataUpdater(client, new Configuration(), Cluster.empty()) { + @Override + public ServerNode getCoordinatorServer() { + return OLD_COORDINATOR; + } + + @Override + public void refreshClusterUntilAvailable() { + throw new AssertionError( + "Non-retriable failures must not refresh metadata"); + } + }; + try (FlussAdmin admin = new FlussAdmin(client, updater)) { + assertThatThrownBy(() -> admin.describeCluster().get(10, TimeUnit.SECONDS)) + .rootCause() + .isSameAs(error); + assertThat(requests).hasValue(1); + } + } + + @Test + void testDiscoveryUsesMetadataWithoutDescribeCluster() throws Exception { + TestingRpcClient client = + new TestingRpcClient( + (node, api) -> { + assertThat(api).isEqualTo(ApiKeys.GET_METADATA); + MetadataResponse response = new MetadataResponse(); + response.setCoordinatorServer() + .setNodeId(0) + .setHost("coordinator") + .setPort(9123); + response.addTabletServer().setNodeId(1).setHost("tablet").setPort(9124); + return CompletableFuture.completedFuture(response); + }); + MetadataUpdater updater = + new MetadataUpdater(client, new Configuration(), Cluster.empty()) { + @Override + public ServerNode getRandomTabletServer() { + return new ServerNode(1, "tablet", 9124, ServerType.TABLET_SERVER); + } + }; + try (FlussAdmin admin = new FlussAdmin(client, updater)) { + assertThat(admin.getServerNodes().get(10, TimeUnit.SECONDS)) + .containsExactly( + new ServerNode(0, "coordinator", 9123, ServerType.COORDINATOR), + new ServerNode(1, "tablet", 9124, ServerType.TABLET_SERVER)); + } + } + + private static CompletableFuture failed(Throwable error) { + CompletableFuture result = new CompletableFuture<>(); + result.completeExceptionally(error); + return result; + } + + private static class TestingRpcClient implements RpcClient { + private final BiFunction> response; + + private TestingRpcClient( + BiFunction> response) { + this.response = response; + } + + @Override + public boolean connect(ServerNode node) { + return true; + } + + @Override + public CompletableFuture disconnect(String serverUid) { + return CompletableFuture.completedFuture(null); + } + + @Override + public boolean isReady(String serverUid) { + return true; + } + + @Override + public CompletableFuture sendRequest( + ServerNode node, ApiKeys apiKey, ApiMessage request) { + return response.apply(node, apiKey); + } + + @Override + public void close() {} + } +} diff --git a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java index 2cdaef69d15..abcc5f8c1e2 100644 --- a/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java +++ b/fluss-client/src/test/java/org/apache/fluss/client/security/acl/FlussAuthorizationITCase.java @@ -53,6 +53,7 @@ import org.apache.fluss.rpc.gateway.CoordinatorGateway; import org.apache.fluss.rpc.gateway.TabletServerGateway; import org.apache.fluss.rpc.messages.ControlledShutdownRequest; +import org.apache.fluss.rpc.messages.DescribeClusterRequest; import org.apache.fluss.rpc.messages.GetKvSnapshotMetadataRequest; import org.apache.fluss.rpc.messages.InitWriterRequest; import org.apache.fluss.rpc.messages.InitWriterResponse; @@ -790,6 +791,39 @@ void testProduceAndConsumer() throws Exception { } } + @Test + void testDescribeClusterAuthorization() throws Exception { + assertThat(guestAdmin.getServerNodes().get()).hasSize(4); + assertThatThrownBy( + () -> + ((FlussAdmin) guestAdmin) + .getAdminGateway() + .describeCluster(new DescribeClusterRequest()) + .get()) + .rootCause() + .isInstanceOf(AuthorizationException.class) + .hasMessageContaining("DESCRIBE") + .hasMessageContaining("CLUSTER"); + assertThatThrownBy(() -> guestAdmin.describeCluster().get()) + .rootCause() + .isInstanceOf(AuthorizationException.class); + rootAdmin + .createAcls( + Collections.singletonList( + new AclBinding( + Resource.cluster(), + new AccessControlEntry( + guestPrincipal, + "*", + OperationType.DESCRIBE, + PermissionType.ALLOW)))) + .all() + .get(); + assertThat(guestAdmin.describeCluster().get().getTabletServers()).hasSize(3); + assertThat(guestAdmin.describeCluster().get().getTabletServerLoads()).hasSize(3); + assertThat(guestAdmin.getServerNodes().get()).hasSize(4); + } + @Test void testDynamicConfigs() throws ExecutionException, InterruptedException { assertThatThrownBy(() -> guestAdmin.describeClusterConfigs().get()) diff --git a/fluss-client/src/test/java/org/apache/fluss/client/utils/MetadataUtilsTest.java b/fluss-client/src/test/java/org/apache/fluss/client/utils/MetadataUtilsTest.java deleted file mode 100644 index 4b2e93b436c..00000000000 --- a/fluss-client/src/test/java/org/apache/fluss/client/utils/MetadataUtilsTest.java +++ /dev/null @@ -1,136 +0,0 @@ -/* - * 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.utils; - -import org.apache.fluss.cluster.ServerNode; -import org.apache.fluss.cluster.ServerType; -import org.apache.fluss.cluster.rebalance.ServerTag; -import org.apache.fluss.exception.FlussRuntimeException; -import org.apache.fluss.exception.UnsupportedVersionException; -import org.apache.fluss.rpc.messages.DescribeClusterResponse; - -import org.junit.jupiter.api.Test; - -import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CompletionException; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Supplier; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatThrownBy; - -/** Tests for {@link MetadataUtils}. */ -class MetadataUtilsTest { - - private final AtomicInteger fallbackCalls = new AtomicInteger(); - - private final Supplier> fallback = - () -> { - fallbackCalls.incrementAndGet(); - return CompletableFuture.completedFuture("fallback"); - }; - - @Test - void testNoFallbackWhenRequestSucceeds() throws Exception { - CompletableFuture result = - MetadataUtils.fallbackIfUnsupported( - () -> CompletableFuture.completedFuture("request"), fallback); - - assertThat(result.get()).isEqualTo("request"); - assertThat(fallbackCalls).hasValue(0); - } - - @Test - void testFallbackWhenServerDoesNotSupportRequest() throws Exception { - // as reported by the RPC client for a server that does not know the API - CompletableFuture unsupported = new CompletableFuture<>(); - unsupported.completeExceptionally( - new CompletionException( - new UnsupportedVersionException( - "The server does not support DESCRIBE_CLUSTER"))); - - assertThat(MetadataUtils.fallbackIfUnsupported(() -> unsupported, fallback).get()) - .isEqualTo("fallback"); - assertThat(fallbackCalls).hasValue(1); - } - - @Test - void testFallbackWhenRequestThrowsUnsupportedVersion() throws Exception { - Supplier> throwing = - () -> { - throw new UnsupportedVersionException("unsupported"); - }; - - assertThat(MetadataUtils.fallbackIfUnsupported(throwing, fallback).get()) - .isEqualTo("fallback"); - assertThat(fallbackCalls).hasValue(1); - } - - @Test - void testOtherFailuresAreNotHiddenByFallback() { - CompletableFuture failed = new CompletableFuture<>(); - failed.completeExceptionally( - new CompletionException(new FlussRuntimeException("coordinator unavailable"))); - - assertThatThrownBy(() -> MetadataUtils.fallbackIfUnsupported(() -> failed, fallback).get()) - .isInstanceOf(ExecutionException.class) - .cause() - .isInstanceOf(FlussRuntimeException.class) - .hasMessage("coordinator unavailable"); - assertThat(fallbackCalls).hasValue(0); - } - - @Test - void testToServerNodes() { - DescribeClusterResponse response = new DescribeClusterResponse(); - response.setCoordinatorServer().setNodeId(0).setHost("coordinator").setPort(9123); - response.addTabletServer().setNodeId(1).setHost("ts-1").setPort(9124).setRack("rack-1"); - response.addTabletServer() - .setNodeId(2) - .setHost("ts-2") - .setPort(9125) - .setRack("rack-2") - .setServerTag(ServerTag.PERMANENT_OFFLINE.value); - // a tag added by a newer server version is reported as no tag - response.addTabletServer().setNodeId(3).setHost("ts-3").setPort(9126).setServerTag(42); - - List serverNodes = MetadataUtils.toServerNodes(response); - - assertThat(serverNodes) - .containsExactly( - new ServerNode(0, "coordinator", 9123, ServerType.COORDINATOR), - new ServerNode(1, "ts-1", 9124, ServerType.TABLET_SERVER, "rack-1"), - new ServerNode(2, "ts-2", 9125, ServerType.TABLET_SERVER, "rack-2"), - new ServerNode(3, "ts-3", 9126, ServerType.TABLET_SERVER, null)); - assertThat(serverNodes) - .extracting(ServerNode::serverTag) - .containsExactly(null, null, ServerTag.PERMANENT_OFFLINE, null); - } - - @Test - void testToServerNodesWithoutCoordinator() { - DescribeClusterResponse response = new DescribeClusterResponse(); - response.addTabletServer().setNodeId(1).setHost("ts-1").setPort(9124); - - assertThat(MetadataUtils.toServerNodes(response)) - .containsExactly(new ServerNode(1, "ts-1", 9124, ServerType.TABLET_SERVER)); - } -} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListServerTagProcedure.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListServerTagProcedure.java index 9859f827f50..3e9986247df 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListServerTagProcedure.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/procedure/ListServerTagProcedure.java @@ -18,7 +18,6 @@ package org.apache.fluss.flink.procedure; import org.apache.fluss.cluster.ServerNode; -import org.apache.fluss.cluster.ServerType; import org.apache.flink.table.annotation.DataTypeHint; import org.apache.flink.table.annotation.ProcedureHint; @@ -31,6 +30,8 @@ /** * Procedure to list server tags of tablet servers. * + *

    Requires DESCRIBE permission on the cluster resource. + * *

    Usage example: * * 

    @@ -41,10 +42,9 @@ public class ListServerTagProcedure extends ProcedureBase {
 
     @ProcedureHint(output = @DataTypeHint("ROW"))
     public Row[] call(ProcedureContext context) throws Exception {
-        List serverNodes = admin.getServerNodes().get();
         List rows = new ArrayList<>();
-        for (ServerNode node : serverNodes) {
-            if (node.serverType() == ServerType.TABLET_SERVER && node.serverTag() != null) {
+        for (ServerNode node : admin.describeCluster().get().getTabletServers().values()) {
+            if (node.serverTag() != null) {
                 rows.add(Row.of(node.id(), node.rack(), node.serverTag().toString()));
             }
         }
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 aad41fec4dc..2c7154456fd 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
@@ -18,6 +18,7 @@
 package org.apache.fluss.flink.sink.testutils;
 
 import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.admin.ClusterDescription;
 import org.apache.fluss.client.admin.ClusterHealth;
 import org.apache.fluss.client.admin.CreateAclsResult;
 import org.apache.fluss.client.admin.DropAclsResult;
@@ -26,7 +27,6 @@
 import org.apache.fluss.client.admin.OffsetSpec;
 import org.apache.fluss.client.admin.ProducerOffsetsResult;
 import org.apache.fluss.client.admin.RegisterResult;
-import org.apache.fluss.client.admin.TabletServerDescription;
 import org.apache.fluss.client.metadata.ActiveKvSnapshots;
 import org.apache.fluss.client.metadata.KvSnapshotMetadata;
 import org.apache.fluss.client.metadata.KvSnapshots;
@@ -362,7 +362,7 @@ public CompletableFuture getClusterHealth() {
     }
 
     @Override
-    public CompletableFuture> describeTabletServers() {
+    public CompletableFuture describeCluster() {
         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 2e87df41d6c..639d6de8a03 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
@@ -24,8 +24,6 @@
 import org.apache.fluss.rpc.messages.DescribeBucketsResponse;
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
-import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.GetClusterHealthRequest;
 import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
 import org.apache.fluss.rpc.messages.GetDatabaseInfoRequest;
@@ -215,13 +213,4 @@ CompletableFuture describeClusterConfigs(
      */
     @RPC(api = ApiKeys.GET_CLUSTER_HEALTH)
     CompletableFuture getClusterHealth(GetClusterHealthRequest request);
-
-    /**
-     * Describe the replica load of each tablet server in the cluster.
-     *
-     * @return the describe tablet servers response.
-     */
-    @RPC(api = ApiKeys.DESCRIBE_TABLET_SERVERS)
-    CompletableFuture describeTabletServers(
-            DescribeTabletServersRequest 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 33fe1a1131d..bcab67bf79d 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
@@ -125,8 +125,7 @@ public enum ApiKeys {
     LIST_CONSUMER_GROUPS(1072, 0, 0, PUBLIC),
     DELETE_CONSUMER_GROUPS(1073, 0, 0, PUBLIC),
     CREATE_TABLE_FROM_LAKE(1074, 0, 0, PUBLIC),
-    DESCRIBE_TABLET_SERVERS(1075, 0, 0, PUBLIC),
-    DESCRIBE_CLUSTER(1076, 0, 0, PUBLIC);
+    DESCRIBE_CLUSTER(1075, 0, 0, PUBLIC);
 
     private static final Map 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 04110a1f84a..4e51bf0862c 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -259,7 +259,10 @@ message DescribeClusterRequest {
 
 message DescribeClusterResponse {
   optional PbServerNode coordinator_server = 1;
+  // Live tablet servers with addresses, racks and server tags.
   repeated PbServerNode tablet_servers = 2;
+  // Also includes offline servers with assigned replicas. Missing loads are not zero counts.
+  repeated PbTabletServerLoad tablet_server_loads = 3;
 }
 
 // produce log request and response
@@ -963,13 +966,6 @@ message GetClusterHealthResponse {
   required int32 status = 5; // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
 }
 
-message DescribeTabletServersRequest {
-}
-
-message DescribeTabletServersResponse {
-  repeated PbTabletServerLoad tablet_servers = 1;
-}
-
 
 // --------------- Inner classes ----------------
 message PbTabletServerLoad {
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 207e43b990e..3d21c9ffce7 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
@@ -31,8 +31,6 @@
 import org.apache.fluss.rpc.messages.DescribeBucketsResponse;
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
-import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.FetchLogRequest;
 import org.apache.fluss.rpc.messages.FetchLogResponse;
 import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerRequest;
@@ -338,10 +336,4 @@ public CompletableFuture getClusterHealth(
             GetClusterHealthRequest request) {
         return null;
     }
-
-    @Override
-    public CompletableFuture describeTabletServers(
-            DescribeTabletServersRequest request) {
-        return null;
-    }
 }
diff --git a/fluss-rust/crates/fluss/src/proto/fluss.rs b/fluss-rust/crates/fluss/src/proto/fluss.rs
index e51b32920ea..c85769170f3 100644
--- a/fluss-rust/crates/fluss/src/proto/fluss.rs
+++ b/fluss-rust/crates/fluss/src/proto/fluss.rs
@@ -320,8 +320,12 @@ pub struct DescribeClusterRequest {}
 pub struct DescribeClusterResponse {
     #[prost(message, optional, tag = "1")]
     pub coordinator_server: ::core::option::Option,
+    /// Live tablet servers with addresses, racks and server tags.
     #[prost(message, repeated, tag = "2")]
     pub tablet_servers: ::prost::alloc::vec::Vec,
+    /// Also includes offline servers with assigned replicas. Missing loads are not zero counts.
+    #[prost(message, repeated, tag = "3")]
+    pub tablet_server_loads: ::prost::alloc::vec::Vec,
 }
 /// produce log request and response
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -1254,13 +1258,6 @@ pub struct GetClusterHealthResponse {
     #[prost(int32, required, tag = "5")]
     pub status: i32,
 }
-#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
-pub struct DescribeTabletServersRequest {}
-#[derive(Clone, PartialEq, ::prost::Message)]
-pub struct DescribeTabletServersResponse {
-    #[prost(message, repeated, tag = "1")]
-    pub tablet_servers: ::prost::alloc::vec::Vec,
-}
 /// --------------- Inner classes ----------------
 #[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct PbTabletServerLoad {
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 345e6aec958..d1435f808bd 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
@@ -102,8 +102,6 @@
 import org.apache.fluss.rpc.messages.DeleteProducerOffsetsResponse;
 import org.apache.fluss.rpc.messages.DescribeClusterRequest;
 import org.apache.fluss.rpc.messages.DescribeClusterResponse;
-import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.DropAclsRequest;
 import org.apache.fluss.rpc.messages.DropAclsResponse;
 import org.apache.fluss.rpc.messages.DropDatabaseRequest;
@@ -216,6 +214,7 @@
 import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -1851,11 +1850,24 @@ public CompletableFuture removeServerTagByRack(
     @Override
     public CompletableFuture describeCluster(
             DescribeClusterRequest request) {
+        if (authorizer != null) {
+            authorizer.authorize(currentSession(), OperationType.DESCRIBE, Resource.cluster());
+        }
         String listenerName = currentListenerName();
-        return CompletableFuture.completedFuture(
-                buildDescribeClusterResponse(
-                        metadataCache.getCoordinatorServer(listenerName),
-                        metadataCache.getAllAliveTabletServersWithTags(listenerName).values()));
+        AccessContextEvent event =
+                new AccessContextEvent<>(
+                        ctx -> {
+                            DescribeClusterResponse response =
+                                    buildDescribeClusterResponse(
+                                            metadataCache.getCoordinatorServer(listenerName),
+                                            metadataCache
+                                                    .getAllAliveTabletServersWithTags(listenerName)
+                                                    .values());
+                            response.addAllTabletServerLoads(computeTabletServerLoads(ctx));
+                            return response;
+                        });
+        eventManagerSupplier.get().put(event);
+        return event.getResultFuture();
     }
 
     @Override
@@ -1961,21 +1973,8 @@ static GetClusterHealthResponse computeClusterHealth(CoordinatorContext ctx) {
         return response;
     }
 
-    @Override
-    public CompletableFuture describeTabletServers(
-            DescribeTabletServersRequest request) {
-        if (authorizer != null) {
-            authorizer.authorize(currentSession(), OperationType.DESCRIBE, Resource.cluster());
-        }
-
-        AccessContextEvent event =
-                new AccessContextEvent<>(CoordinatorService::computeTabletServerLoads);
-        eventManagerSupplier.get().put(event);
-        return event.getResultFuture();
-    }
-
     @VisibleForTesting
-    static DescribeTabletServersResponse computeTabletServerLoads(CoordinatorContext ctx) {
+    static Collection computeTabletServerLoads(CoordinatorContext ctx) {
         // sorted by server id for a deterministic response order
         Map loads = new TreeMap<>();
         // report live and shutting-down servers even if they host no replicas, so that
@@ -1988,9 +1987,7 @@ static DescribeTabletServersResponse computeTabletServerLoads(CoordinatorContext
             countBucketReplicas(ctx, tb, loads);
         }
 
-        DescribeTabletServersResponse response = new DescribeTabletServersResponse();
-        response.addAllTabletServers(loads.values());
-        return response;
+        return loads.values();
     }
 
     private static void countBucketReplicas(
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 5ce0a277d6e..d49f9c058b2 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
@@ -50,8 +50,6 @@
 import org.apache.fluss.rpc.messages.DeleteConsumerGroupsResponse;
 import org.apache.fluss.rpc.messages.DeleteConsumerOffsetsRequest;
 import org.apache.fluss.rpc.messages.DeleteConsumerOffsetsResponse;
-import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.FetchLogRequest;
 import org.apache.fluss.rpc.messages.FetchLogResponse;
 import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerRequest;
@@ -882,29 +880,6 @@ public CompletableFuture deleteConsumerGroups(
         return response;
     }
 
-    @Override
-    public CompletableFuture describeTabletServers(
-            DescribeTabletServersRequest request) {
-        // Tablet servers don't own the cluster-wide replica 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 callers want.
-            CompletableFuture failed = new CompletableFuture<>();
-            failed.completeExceptionally(
-                    new StaleMetadataException(
-                            "Tablet server has not yet received coordinator metadata; tablet"
-                                    + " server loads are unavailable."));
-            return failed;
-        }
-        return coordinatorGateway.describeTabletServers(request);
-    }
-
     @Override
     public CompletableFuture scanKv(ScanKvRequest request) {
         ScanKvResponse response = new ScanKvResponse();
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/DescribeTabletServersTest.java b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TabletServerLoadTest.java
similarity index 89%
rename from fluss-server/src/test/java/org/apache/fluss/server/coordinator/DescribeTabletServersTest.java
rename to fluss-server/src/test/java/org/apache/fluss/server/coordinator/TabletServerLoadTest.java
index dc38190ca3a..194a677c477 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/DescribeTabletServersTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/TabletServerLoadTest.java
@@ -20,7 +20,6 @@
 import org.apache.fluss.cluster.Endpoint;
 import org.apache.fluss.cluster.ServerType;
 import org.apache.fluss.metadata.TableBucket;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.GetClusterHealthResponse;
 import org.apache.fluss.rpc.messages.PbTabletServerLoad;
 import org.apache.fluss.server.metadata.ServerInfo;
@@ -31,12 +30,13 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for {@link CoordinatorService#computeTabletServerLoads(CoordinatorContext)}. */
-class DescribeTabletServersTest {
+class TabletServerLoadTest {
 
     private CoordinatorContext ctx;
 
@@ -49,11 +49,9 @@ void setUp() {
 
     @Test
     void testEmptyClusterReportsLiveServersWithZeroCounters() {
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
 
-        assertThat(resp.getTabletServersList())
-                .extracting(PbTabletServerLoad::getServerId)
-                .containsExactly(0, 1, 2);
+        assertThat(resp).extracting(PbTabletServerLoad::getServerId).containsExactly(0, 1, 2);
         assertLoad(resp, 0, 0, 0, 0, 0);
         assertLoad(resp, 1, 0, 0, 0, 0);
         assertLoad(resp, 2, 0, 0, 0, 0);
@@ -66,7 +64,7 @@ void testAllInSyncAllLeadersActive() {
         ctx.putBucketLeaderAndIsr(
                 tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1, 2), Collections.emptyList(), 0, 1));
 
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
 
         assertLoad(resp, 0, 1, 1, 1, 1);
         assertLoad(resp, 1, 1, 1, 0, 0);
@@ -81,7 +79,7 @@ void testReplicaOutOfIsrOnlyAffectsThatServer() {
         ctx.putBucketLeaderAndIsr(
                 tb, new LeaderAndIsr(0, 1, Arrays.asList(0, 1), Collections.emptyList(), 0, 1));
 
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
 
         assertLoad(resp, 0, 1, 1, 1, 1);
         assertLoad(resp, 1, 1, 1, 0, 0);
@@ -121,7 +119,7 @@ void testNoLeaderBucketAttributesLeadershipToNoServer() {
                         0,
                         1));
 
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
 
         assertLoad(resp, 0, 1, 1, 0, 0, 1);
         assertLoad(resp, 1, 1, 1, 0, 0, 1);
@@ -133,7 +131,7 @@ void testBucketWithoutLeaderAndIsrIsLeaderless() {
         TableBucket tb = new TableBucket(1L, 0);
         ctx.updateBucketReplicaAssignment(tb, Arrays.asList(0, 1));
 
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
 
         assertLoad(resp, 0, 1, 0, 0, 0, 1);
         assertLoad(resp, 1, 1, 0, 0, 0, 1);
@@ -172,7 +170,7 @@ void testPartitionedTableBucket() {
                 new LeaderAndIsr(
                         0, 1, Collections.singletonList(0), Collections.emptyList(), 0, 1));
 
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
 
         assertLoad(resp, 0, 1, 1, 1, 1);
         assertLoad(resp, 1, 1, 0, 0, 0);
@@ -198,14 +196,14 @@ void testPerServerSumsReconcileWithClusterHealth() {
                 tb3, new LeaderAndIsr(0, 1, Arrays.asList(0, 2), Collections.emptyList(), 0, 1));
         ctx.addPendingLeaderActivation(tb3);
 
-        DescribeTabletServersResponse resp = CoordinatorService.computeTabletServerLoads(ctx);
+        Collection resp = CoordinatorService.computeTabletServerLoads(ctx);
         GetClusterHealthResponse health = CoordinatorService.computeClusterHealth(ctx);
 
         int numReplicas = 0;
         int inSyncReplicas = 0;
         int numLeaderReplicas = 0;
         int activeLeaderReplicas = 0;
-        for (PbTabletServerLoad load : resp.getTabletServersList()) {
+        for (PbTabletServerLoad load : resp) {
             numReplicas += load.getNumReplicas();
             inSyncReplicas += load.getInSyncReplicas();
             numLeaderReplicas += load.getNumLeaderReplicas();
@@ -220,7 +218,7 @@ void testPerServerSumsReconcileWithClusterHealth() {
     }
 
     private static void assertLoad(
-            DescribeTabletServersResponse resp,
+            Collection resp,
             int serverId,
             int numReplicas,
             int inSyncReplicas,
@@ -237,7 +235,7 @@ private static void assertLoad(
     }
 
     private static void assertLoad(
-            DescribeTabletServersResponse resp,
+            Collection resp,
             int serverId,
             int numReplicas,
             int inSyncReplicas,
@@ -264,7 +262,7 @@ private static void assertLoad(
 
     private void assertServersGreenIffClusterGreen() {
         boolean allServersGreen =
-                CoordinatorService.computeTabletServerLoads(ctx).getTabletServersList().stream()
+                CoordinatorService.computeTabletServerLoads(ctx).stream()
                         .allMatch(
                                 l ->
                                         l.getInSyncReplicas() == l.getNumReplicas()
@@ -276,8 +274,8 @@ private void assertServersGreenIffClusterGreen() {
         assertThat(allServersGreen).isEqualTo(clusterGreen);
     }
 
-    private static PbTabletServerLoad findLoad(DescribeTabletServersResponse resp, int serverId) {
-        return resp.getTabletServersList().stream()
+    private static PbTabletServerLoad findLoad(Collection resp, int serverId) {
+        return resp.stream()
                 .filter(l -> l.getServerId() == serverId)
                 .findFirst()
                 .orElseThrow(
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 e63f7a5982c..867a43d8aa6 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
@@ -68,8 +68,6 @@
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
 import org.apache.fluss.rpc.messages.DescribeClusterRequest;
 import org.apache.fluss.rpc.messages.DescribeClusterResponse;
-import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.DropAclsRequest;
 import org.apache.fluss.rpc.messages.DropAclsResponse;
 import org.apache.fluss.rpc.messages.DropDatabaseRequest;
@@ -521,12 +519,6 @@ public CompletableFuture getClusterHealth(
         throw new UnsupportedOperationException();
     }
 
-    @Override
-    public CompletableFuture describeTabletServers(
-            DescribeTabletServersRequest request) {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public CompletableFuture describeCluster(
             DescribeClusterRequest 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 d48ca4c94b4..36c95e55e73 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
@@ -36,8 +36,6 @@
 import org.apache.fluss.rpc.messages.DescribeBucketsResponse;
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsRequest;
 import org.apache.fluss.rpc.messages.DescribeClusterConfigsResponse;
-import org.apache.fluss.rpc.messages.DescribeTabletServersRequest;
-import org.apache.fluss.rpc.messages.DescribeTabletServersResponse;
 import org.apache.fluss.rpc.messages.FetchLogRequest;
 import org.apache.fluss.rpc.messages.FetchLogResponse;
 import org.apache.fluss.rpc.messages.FindConsumerGroupOwnerRequest;
@@ -260,12 +258,6 @@ public CompletableFuture getClusterHealth(
         throw new UnsupportedOperationException();
     }
 
-    @Override
-    public CompletableFuture describeTabletServers(
-            DescribeTabletServersRequest request) {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public CompletableFuture databaseExists(DatabaseExistsRequest request) {
         throw new UnsupportedOperationException();
diff --git a/website/docs/apis/java/index.md b/website/docs/apis/java/index.md
index 74c2bfa44c2..83719b68b2b 100644
--- a/website/docs/apis/java/index.md
+++ b/website/docs/apis/java/index.md
@@ -144,6 +144,31 @@ TableInfo tableInfo = admin.getTableInfo(tablePath).get(); // blocking call
 System.out.println(tableInfo);
 ```
 
+## Cluster Inspection
+
+Since Fluss 1.1, `Admin.describeCluster()` returns a `ClusterDescription` containing the coordinator,
+live tablet server nodes (addresses, racks and server tags), and replica loads. Both tablet server
+maps are keyed by server ID:
+
+```java
+ClusterDescription cluster = admin.describeCluster().get();
+int serverId = 1;
+ServerNode node = cluster.getTabletServers().get(serverId);
+TabletServerDescription load = cluster.getTabletServerLoads().get(serverId);
+boolean hasNoAssignedReplicas = load != null && load.getNumReplicas() == 0;
+```
+
+The load map also includes offline servers that still have assigned replicas, so a server may have
+a load entry without a live node entry. Live servers with no replicas have explicit zero counts.
+A missing load entry means statistics are unavailable; it must not be treated as proof that a
+server has been evacuated.
+
+This API requires `DESCRIBE` permission on the cluster and refreshes metadata and retries once on
+retriable failures, including coordinator address changes. Servers that do not support it return
+`UnsupportedVersionException`. For basic discovery, `Admin.getServerNodes()` continues to return
+the coordinator and live tablet servers without requiring cluster `DESCRIBE` permission; it does
+not include server tags or replica loads.
+
 ## Table Statistics
 
 The Admin API provides the `getTableStats` method to retrieve statistics about a table, such as the current row count.
diff --git a/website/docs/engine-flink/procedures.md b/website/docs/engine-flink/procedures.md
index 870227f0d50..01d85f28fe6 100644
--- a/website/docs/engine-flink/procedures.md
+++ b/website/docs/engine-flink/procedures.md
@@ -536,7 +536,8 @@ This procedure takes no parameters.
 **Important Notes:**
 
 - Only live TabletServers that have a tag assigned are included in the result. If no TabletServer has a tag, an empty result is returned.
-- The tags are read from the Coordinator. Against a cluster older than Fluss 1.1, which cannot report server tags, the result is always empty.
+- The caller must have `DESCRIBE` permission on the cluster.
+- The tags are read from the Coordinator through `Admin.describeCluster()`. This requires Fluss 1.1 or later; older clusters fail with `UnsupportedVersionException`.
 
 **Example:**
 
diff --git a/website/docs/security/authorization.md b/website/docs/security/authorization.md
index 0287bbe23bf..2c2cb6e61d5 100644
--- a/website/docs/security/authorization.md
+++ b/website/docs/security/authorization.md
@@ -91,7 +91,7 @@ Below is a summary of the currently public protocols and their relationship with
 | TABLE_EXISTS | DESCRIBE | Table | Returns `false` if the user lacks permission. |
 | LIST_TABLES | DESCRIBE | Table | Only tables that the user has permission to access are returned. |
 | LIST_PARTITION_INFOS | DESCRIBE | Table | Requires permission on the owning table. Partition-level ACLs are not checked. |
-| GET_METADATA | DESCRIBE | Table | Only metadata that the user has permission to access is returned. |
+| GET_METADATA | DESCRIBE | Table | Only table metadata that the user has permission to access is returned. Server discovery through `Admin.getServerNodes()` does not require cluster `DESCRIBE` permission and does not return tags or replica loads. |
 | GET_LATEST_KV_SNAPSHOTS | DESCRIBE | Table | |
 | GET_KV_SNAPSHOT_METADATA | DESCRIBE | Table | |
 | GET_LAKE_SNAPSHOT | DESCRIBE | Table | |
@@ -122,6 +122,7 @@ Below is a summary of the currently public protocols and their relationship with
 | LIST_REBALANCE_PROGRESS | DESCRIBE | Cluster | |
 | CANCEL_REBALANCE | WRITE | Cluster | |
 | GET_CLUSTER_HEALTH | DESCRIBE | Cluster | |
+| DESCRIBE_CLUSTER | DESCRIBE | Cluster | Returns live server nodes with tags and replica loads, including offline servers with assigned replicas. |
 | REGISTER_PRODUCER_OFFSETS | WRITE | Table | Requires `WRITE` on all tables in the request. |
 | GET_PRODUCER_OFFSETS | READ | Table | Offsets for tables without `READ` permission are filtered out. |
 | DELETE_PRODUCER_OFFSETS | WRITE | Table | Requires `WRITE` on all tables in the producer offset snapshot. |

From a4faeff8a643f609221fdcd8101e33649aac75a3 Mon Sep 17 00:00:00 2001
From: yunhong <337361684@qq.com>
Date: Tue, 27 Oct 2026 10:30:16 +0800
Subject: [PATCH 2/2] [hotfix] Make NotCoordinatorLeaderException retriable

Classify NotCoordinatorLeaderException as InvalidMetadataException so
retryable calls refresh cluster metadata before retrying after a leader
change. Cover successful recovery and the one-retry limit.

Co-Authored-By: Codex 
AI-Model: gpt-6
---
 .../fluss/client/admin/FlussAdminTest.java     | 18 +++++++++++++-----
 .../NotCoordinatorLeaderException.java         |  7 +++++--
 2 files changed, 18 insertions(+), 7 deletions(-)

diff --git a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java
index 6a40279aaea..10d8fb1e401 100644
--- a/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java
+++ b/fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminTest.java
@@ -24,16 +24,17 @@
 import org.apache.fluss.cluster.ServerType;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.AuthorizationException;
-import org.apache.fluss.exception.NetworkException;
 import org.apache.fluss.exception.UnsupportedVersionException;
 import org.apache.fluss.rpc.RpcClient;
 import org.apache.fluss.rpc.messages.ApiMessage;
 import org.apache.fluss.rpc.messages.DescribeClusterResponse;
 import org.apache.fluss.rpc.messages.MetadataResponse;
 import org.apache.fluss.rpc.protocol.ApiKeys;
+import org.apache.fluss.rpc.protocol.Errors;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.util.ArrayList;
@@ -55,8 +56,15 @@ class FlussAdminTest {
             new ServerNode(0, "new-coordinator", 9123, ServerType.COORDINATOR);
 
     @ParameterizedTest
-    @ValueSource(booleans = {true, false})
-    void testDescribeClusterRefreshesAndRetriesOnce(boolean retrySucceeds) throws Exception {
+    @CsvSource({
+        "NETWORK_EXCEPTION, true",
+        "NETWORK_EXCEPTION, false",
+        "NOT_COORDINATOR_LEADER_EXCEPTION, true",
+        "NOT_COORDINATOR_LEADER_EXCEPTION, false"
+    })
+    void testDescribeClusterRefreshesAndRetriesOnce(Errors error, boolean retrySucceeds)
+            throws Exception {
+        RuntimeException failure = error.exception("coordinator unavailable");
         AtomicInteger refreshes = new AtomicInteger();
         List contactedNodes = new ArrayList<>();
         TestingRpcClient client =
@@ -65,7 +73,7 @@ void testDescribeClusterRefreshesAndRetriesOnce(boolean retrySucceeds) throws Ex
                             assertThat(api).isEqualTo(ApiKeys.DESCRIBE_CLUSTER);
                             contactedNodes.add(node);
                             if (node.equals(OLD_COORDINATOR) || !retrySucceeds) {
-                                return failed(new NetworkException("coordinator unavailable"));
+                                return failed(failure);
                             }
                             DescribeClusterResponse response = new DescribeClusterResponse();
                             response.setCoordinatorServer()
@@ -93,7 +101,7 @@ public void refreshClusterUntilAvailable() {
             } else {
                 assertThatThrownBy(() -> admin.describeCluster().get(10, TimeUnit.SECONDS))
                         .rootCause()
-                        .isInstanceOf(NetworkException.class);
+                        .isSameAs(failure);
             }
             assertThat(contactedNodes).containsExactly(OLD_COORDINATOR, NEW_COORDINATOR);
             assertThat(refreshes).hasValue(1);
diff --git a/fluss-common/src/main/java/org/apache/fluss/exception/NotCoordinatorLeaderException.java b/fluss-common/src/main/java/org/apache/fluss/exception/NotCoordinatorLeaderException.java
index faa2f55876c..c8da8d219e8 100644
--- a/fluss-common/src/main/java/org/apache/fluss/exception/NotCoordinatorLeaderException.java
+++ b/fluss-common/src/main/java/org/apache/fluss/exception/NotCoordinatorLeaderException.java
@@ -18,11 +18,14 @@
 package org.apache.fluss.exception;
 
 /**
- * Exception thrown when a request is sent to a stand by coordinator server.
+ * Exception thrown when a request is sent to a standby coordinator server.
+ *
+ * 
    The server rejects the request before executing it. Clients should refresh cluster metadata to
+ * discover the current coordinator leader before retrying.
  *
  * @since 1.0
  */
-public class NotCoordinatorLeaderException extends ApiException {
+public class NotCoordinatorLeaderException extends InvalidMetadataException {
 
     private static final long serialVersionUID = 1L;