[client] Unify cluster inspection in DescribeCluster

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.
-     *
-     * <p>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<List<ServerNode>> getServerNodes();
 
@@ -929,10 +926,21 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
     CompletableFuture<ClusterHealth> 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.
+     *
+     * <p>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.
+     *
+     * <p>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}.
      *
-     * <p>The returned list contains {@link TabletServerDescription} counters per tablet server.
-     * Live tablet servers are always included, even when they host no replicas.
+     * <p>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.
      *
      * <p>This API is designed for operational tooling (e.g., a Kubernetes operator):
      *
@@ -946,10 +954,10 @@ CompletableFuture<RegisterResult> registerProducerOffsets(
      *   <li>reporting per-server tablet load in cluster status.
      * </ul>
      *
-     * @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<List<TabletServerDescription>> describeTabletServers();
+    CompletableFuture<ClusterDescription> describeCluster();
 
     /**
      * List per-bucket remote log manifest entries for a table or partition scope.

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()}.
+ *
+ * <p>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<Integer, ServerNode> tabletServers;
+    private final Map<Integer, TabletServerDescription> tabletServerLoads;
+
+    /** Creates a cluster description with defensive copies of the server maps. */
+    public ClusterDescription(
+            @Nullable ServerNode coordinatorServer,
+            Map<Integer, ServerNode> tabletServers,
+            Map<Integer, TabletServerDescription> 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<ServerNode> getCoordinatorServer() {
+        return Optional.ofNullable(coordinatorServer);
+    }
+
+    /** Returns the live tablet servers by server ID, in an immutable map ordered by ID. */
+    public Map<Integer, ServerNode> getTabletServers() {
+        return tabletServers;
+    }
+
+    /**
+     * Returns the reported replica loads by server ID, in an immutable map ordered by ID.
+     *
+     * <p>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<Integer, TabletServerDescription> 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
+                + '}';
+    }
+}

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<List<ServerNode>> 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<List<ServerNode>> getServerNodesFromMetadata() {
         CompletableFuture<List<ServerNode>> future = new CompletableFuture<>();
         CompletableFuture.runAsync(
                 () -> {
@@ -1211,9 +1200,10 @@ public CompletableFuture<ClusterHealth> getClusterHealth() {
     }
 
     @Override
-    public CompletableFuture<List<TabletServerDescription>> describeTabletServers() {
-        return gateway.describeTabletServers(new DescribeTabletServersRequest())
-                .thenApply(ClientRpcMessageUtils::toTabletServerDescriptions);
+    public CompletableFuture<ClusterDescription> describeCluster() {
+        return retryableCoordinatorGateway
+                .describeCluster(new DescribeClusterRequest())
+                .thenApply(ClientRpcMessageUtils::toClusterDescription);
     }
 
     @VisibleForTesting

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()}.
  *
  * <p>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.

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<ReadyWriteBatch> readyBatches) {
         ProduceLogRequest request =
@@ -987,18 +998,50 @@ private static ClusterHealthStatus toClusterHealthStatus(int pbStatus) {
         }
     }
 
-    public static List<TabletServerDescription> 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<Integer, ServerNode> tabletServers = new HashMap<>();
+        for (PbServerNode node : response.getTabletServersList()) {
+            tabletServers.put(node.getNodeId(), toServerNode(node, ServerType.TABLET_SERVER));
+        }
+        Map<Integer, TabletServerDescription> 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;
+        }
     }
 }