[client] Expose all coordinator servers (leader and standbys) via the client API

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

@@ -23,6 +23,7 @@
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.client.metadata.RemoteLogManifestInfo;
+import org.apache.fluss.cluster.CoordinatorServerInfo;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.rebalance.GoalType;
 import org.apache.fluss.cluster.rebalance.RebalanceProgress;
@@ -93,6 +94,20 @@ public interface Admin extends AutoCloseable {
     /** Get the current server node information. asynchronously. */
     CompletableFuture<List<ServerNode>> getServerNodes();
 
+    /**
+     * Get all coordinator servers with role and liveness information asynchronously.
+     *
+     * <p>This returns all coordinators (leader and standby) registered in the cluster with their
+     * current role and liveness status. This is useful for monitoring coordinator HA topology and
+     * implementing readiness checks for standby coordinators.
+     *
+     * @return A list of {@link CoordinatorServerInfo} containing coordinator id, node, role
+     *     (LEADER/STANDBY), and liveness information. Returns empty list if no coordinators are
+     *     registered.
+     * @since 1.0
+     */
+    CompletableFuture<List<CoordinatorServerInfo>> describeCoordinators();
+
     /**
      * Get the latest table schema of the given table asynchronously.
      *

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

@@ -27,6 +27,7 @@
 import org.apache.fluss.client.utils.ClientRpcMessageUtils;
 import org.apache.fluss.client.utils.ClientUtils;
 import org.apache.fluss.cluster.Cluster;
+import org.apache.fluss.cluster.CoordinatorServerInfo;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.rebalance.GoalType;
 import org.apache.fluss.cluster.rebalance.RebalanceProgress;
@@ -215,6 +216,33 @@ public CompletableFuture<List<ServerNode>> getServerNodes() {
         return future;
     }
 
+    @Override
+    public CompletableFuture<List<CoordinatorServerInfo>> describeCoordinators() {
+        CompletableFuture<List<CoordinatorServerInfo>> future = new CompletableFuture<>();
+        CompletableFuture.runAsync(
+                () -> {
+                    try {
+                        Cluster cluster =
+                                sendMetadataRequestAndRebuildCluster(
+                                        readOnlyGateway,
+                                        false,
+                                        metadataUpdater.getCluster(),
+                                        null,
+                                        null,
+                                        null);
+                        List<CoordinatorServerInfo> coordinatorServerInfos =
+                                cluster.getCoordinatorServerInfos();
+                        future.complete(
+                                coordinatorServerInfos != null
+                                        ? coordinatorServerInfos
+                                        : new ArrayList<>());
+                    } catch (Throwable t) {
+                        future.completeExceptionally(t);
+                    }
+                });
+        return future;
+    }
+
     @Override
     public CompletableFuture<SchemaInfo> getTableSchema(TablePath tablePath) {
         GetTableSchemaRequest request = new GetTableSchemaRequest();

fluss-client/src/main/java/org/apache/fluss/client/utils/MetadataUtils.java

@@ -19,6 +19,8 @@
 
 import org.apache.fluss.cluster.BucketLocation;
 import org.apache.fluss.cluster.Cluster;
+import org.apache.fluss.cluster.CoordinatorRole;
+import org.apache.fluss.cluster.CoordinatorServerInfo;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.ServerType;
 import org.apache.fluss.exception.StaleMetadataException;
@@ -32,6 +34,7 @@
 import org.apache.fluss.rpc.messages.MetadataRequest;
 import org.apache.fluss.rpc.messages.MetadataResponse;
 import org.apache.fluss.rpc.messages.PbBucketMetadata;
+import org.apache.fluss.rpc.messages.PbCoordinatorServerInfo;
 import org.apache.fluss.rpc.messages.PbPartitionMetadata;
 import org.apache.fluss.rpc.messages.PbServerNode;
 import org.apache.fluss.rpc.messages.PbTableMetadata;
@@ -118,6 +121,8 @@ public static Cluster sendMetadataRequestAndRebuildCluster(
                                 throw new StaleMetadataException("Alive tablet server is empty.");
                             }
                             ServerNode coordinatorServer = getCoordinatorServer(response);
+                            List<CoordinatorServerInfo> coordinatorServerInfos =
+                                    getCoordinatorServerInfos(response);
 
                             Map<TablePath, Long> newTablePathToTableId;
                             Map<PhysicalTablePath, List<BucketLocation>> newBucketLocations;
@@ -159,6 +164,7 @@ public static Cluster sendMetadataRequestAndRebuildCluster(
                             return new Cluster(
                                     newAliveTabletServers,
                                     coordinatorServer,
+                                    coordinatorServerInfos,
                                     newBucketLocations,
                                     newTablePathToTableId,
                                     newPartitionIdByPath,
@@ -309,6 +315,33 @@ private static ServerNode getCoordinatorServer(MetadataResponse response) {
         }
     }
 
+    @Nullable
+    private static List<CoordinatorServerInfo> getCoordinatorServerInfos(
+            MetadataResponse response) {
+        List<PbCoordinatorServerInfo> pbCoordinatorServerInfos =
+                response.getCoordinatorServersList();
+        if (pbCoordinatorServerInfos.isEmpty()) {
+            return null;
+        }
+
+        List<CoordinatorServerInfo> coordinatorServerInfos = new ArrayList<>();
+        for (PbCoordinatorServerInfo pbCoordinatorServerInfo : pbCoordinatorServerInfos) {
+            int nodeId = pbCoordinatorServerInfo.getId();
+            ServerNode node =
+                    new ServerNode(
+                            nodeId,
+                            pbCoordinatorServerInfo.getCoordinatorServer().getHost(),
+                            pbCoordinatorServerInfo.getCoordinatorServer().getPort(),
+                            ServerType.COORDINATOR);
+            CoordinatorRole role = CoordinatorRole.fromRoleId(pbCoordinatorServerInfo.getRole());
+            boolean isAlive = pbCoordinatorServerInfo.isIsAlive();
+            String coordinatorId = node.uid();
+            coordinatorServerInfos.add(
+                    new CoordinatorServerInfo(coordinatorId, node, role, isAlive));
+        }
+        return coordinatorServerInfos;
+    }
+
     private static Map<Integer, ServerNode> getAliveTabletServers(MetadataResponse response) {
         Map<Integer, ServerNode> aliveTabletServers = new HashMap<>();
         response.getTabletServersList()

fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java

@@ -49,6 +49,7 @@
 @Internal
 public final class Cluster {
     @Nullable private final ServerNode coordinatorServer;
+    @Nullable private final List<CoordinatorServerInfo> coordinatorServerInfos;
     private final Map<PhysicalTablePath, List<BucketLocation>> availableLocationsByPath;
     private final Map<TableBucket, BucketLocation> availableLocationByBucket;
     private final Map<Integer, ServerNode> aliveTabletServersById;
@@ -66,7 +67,29 @@ public Cluster(
             Map<TablePath, Long> tableIdByPath,
             Map<PhysicalTablePath, Long> partitionsIdByPath,
             Map<TableOrPartition, Integer> bucketCountByTableOrPartition) {
+        this(
+                aliveTabletServersById,
+                coordinatorServer,
+                null,
+                bucketLocationsByPath,
+                tableIdByPath,
+                partitionsIdByPath,
+                bucketCountByTableOrPartition);
+    }
+
+    public Cluster(
+            Map<Integer, ServerNode> aliveTabletServersById,
+            @Nullable ServerNode coordinatorServer,
+            @Nullable List<CoordinatorServerInfo> coordinatorServerInfos,
+            Map<PhysicalTablePath, List<BucketLocation>> bucketLocationsByPath,
+            Map<TablePath, Long> tableIdByPath,
+            Map<PhysicalTablePath, Long> partitionsIdByPath,
+            Map<TableOrPartition, Integer> bucketCountByTableOrPartition) {
         this.coordinatorServer = coordinatorServer;
+        this.coordinatorServerInfos =
+                coordinatorServerInfos != null
+                        ? Collections.unmodifiableList(new ArrayList<>(coordinatorServerInfos))
+                        : null;
         this.aliveTabletServersById = Collections.unmodifiableMap(aliveTabletServersById);
         this.aliveTabletServers =
                 Collections.unmodifiableList(new ArrayList<>(aliveTabletServersById.values()));
@@ -161,6 +184,7 @@ public Cluster invalidPhysicalTableBucketMeta(Set<PhysicalTablePath> physicalTab
         return new Cluster(
                 new HashMap<>(aliveTabletServersById),
                 coordinatorServer,
+                coordinatorServerInfos,
                 newBucketLocationsByPath,
                 new HashMap<>(tableIdByPath),
                 new HashMap<>(partitionsIdByPath),
@@ -178,6 +202,7 @@ public Cluster invalidPhysicalTableBucketAndPartitionMeta(
         return new Cluster(
                 new HashMap<>(aliveTabletServersById),
                 coordinatorServer,
+                coordinatorServerInfos,
                 new HashMap<>(cluster.availableLocationsByPath),
                 new HashMap<>(tableIdByPath),
                 newPartitionsIdByPath,
@@ -189,6 +214,11 @@ public ServerNode getCoordinatorServer() {
         return coordinatorServer;
     }
 
+    @Nullable
+    public List<CoordinatorServerInfo> getCoordinatorServerInfos() {
+        return coordinatorServerInfos;
+    }
+
     /**
      * @return The known set of alive tablet servers.
      */

fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorRole.java

@@ -0,0 +1,45 @@
+/*
+ * 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.cluster;
+
+/** The role of the coordinator server in the Fluss cluster. */
+public enum CoordinatorRole {
+    LEADER(0),
+    STANDBY(1),
+    UNKNOWN(-1);
+
+    private final int value;
+
+    CoordinatorRole(int value) {
+        this.value = value;
+    }
+
+    public static CoordinatorRole fromRoleId(int value) {
+        for (CoordinatorRole role : CoordinatorRole.values()) {
+            if (role.getValue() == value) {
+                return role;
+            }
+        }
+        return UNKNOWN;
+    }
+
+    public int getValue() {
+        return value;
+    }
+}

fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorServerInfo.java

@@ -0,0 +1,66 @@
+/*
+ * 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.cluster;
+
+/** The information of the coordinator server in Fluss cluster. */
+public class CoordinatorServerInfo {
+    private final String coordinatorId;
+    private final ServerNode node;
+    private final CoordinatorRole role;
+    private final boolean isAlive;
+
+    public CoordinatorServerInfo(
+            String coordinatorId, ServerNode node, CoordinatorRole role, boolean isAlive) {
+        this.coordinatorId = coordinatorId;
+        this.node = node;
+        this.role = role;
+        this.isAlive = isAlive;
+    }
+
+    public String getCoordinatorId() {
+        return coordinatorId;
+    }
+
+    public ServerNode getNode() {
+        return node;
+    }
+
+    public CoordinatorRole getRole() {
+        return role;
+    }
+
+    public boolean isAlive() {
+        return isAlive;
+    }
+
+    @Override
+    public String toString() {
+        return "CoordinatorServerInfo{"
+                + "coordinatorId='"
+                + coordinatorId
+                + '\''
+                + ", node="
+                + node
+                + ", role="
+                + role
+                + ", isAlive="
+                + isAlive
+                + '}';
+    }
+}

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

@@ -31,6 +31,7 @@
 import org.apache.fluss.client.metadata.KvSnapshots;
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.client.metadata.RemoteLogManifestInfo;
+import org.apache.fluss.cluster.CoordinatorServerInfo;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.rebalance.GoalType;
 import org.apache.fluss.cluster.rebalance.RebalanceProgress;
@@ -89,6 +90,11 @@ public CompletableFuture<List<ServerNode>> getServerNodes() {
         throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
     }
 
+    @Override
+    public CompletableFuture<List<CoordinatorServerInfo>> describeCoordinators() {
+        throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");
+    }
+
     @Override
     public CompletableFuture<SchemaInfo> getTableSchema(TablePath tablePath) {
         throw new UnsupportedOperationException("Not implemented in TestAdminAdapter");

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

@@ -215,11 +215,19 @@ message MetadataRequest {
   repeated int64 partitions_id = 3 [packed = true];
 }
 
+message PbCoordinatorServerInfo{
+    required int32 id = 1;
+    required PbServerNode coordinator_server = 2;
+    required int32 role = 3;
+    required bool isAlive = 4;
+}
+
 message MetadataResponse {
   optional PbServerNode coordinator_server = 1;
   repeated PbServerNode tablet_servers = 2;
   repeated PbTableMetadata table_metadata = 3;
   repeated PbPartitionMetadata partition_metadata = 4;
+    repeated  PbCoordinatorServerInfo coordinator_servers = 5;
 }
 
 // update metadata request and response, only send between server.