From cfb4adb47b8c0b699856a3a1015daec64c61ebca Mon Sep 17 00:00:00 2001 From: MaheshS08 Date: Thu, 17 Sep 2026 01:48:42 +0530 Subject: [PATCH 1/3] [client] Expose all coordinator servers (leader and standbys) via the client API --- .../org/apache/fluss/client/admin/Admin.java | 15 ++++ .../apache/fluss/client/admin/FlussAdmin.java | 49 ++++++++++++ .../fluss/client/utils/MetadataUtils.java | 33 ++++++++ .../org/apache/fluss/cluster/Cluster.java | 30 +++++++ .../apache/fluss/cluster/CoordinatorRole.java | 45 +++++++++++ .../fluss/cluster/CoordinatorServerInfo.java | 66 ++++++++++++++++ .../sink/testutils/TestAdminAdapter.java | 6 ++ fluss-rpc/src/main/proto/FlussApi.proto | 8 ++ .../apache/fluss/server/RpcServiceBase.java | 79 ++++++++++++++++++- .../server/utils/ServerRpcMessageUtils.java | 36 +++++++++ .../fluss/server/zk/ZooKeeperClient.java | 40 ++++++++++ .../fluss/server/zk/ZooKeeperClientTest.java | 43 ++++++++++ 12 files changed, 449 insertions(+), 1 deletion(-) create mode 100644 fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorRole.java create mode 100644 fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorServerInfo.java 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 8ec899a998..a5bd2fb8cd 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 @@ -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; @@ -92,6 +93,20 @@ public interface Admin extends AutoCloseable { /** Get the current server node information. asynchronously. */ CompletableFuture> getServerNodes(); + /** + * Get all coordinator servers with role and liveness information asynchronously. + * + *

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> describeCoordinators(); + /** * Get the latest table schema of the given table asynchronously. * 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 d9251f4fc0..5f5207b817 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 @@ -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; @@ -199,6 +200,27 @@ public CompletableFuture> getServerNodes() { null, null, null); + // Add all coordinator servers (leader + standbys) + // List coordinatorServerInfos + // = + // cluster.getCoordinatorServerInfos(); + // if (coordinatorServerInfos != null && + // !coordinatorServerInfos.isEmpty()) { + // for (CoordinatorServerInfo + // coordinatorServerInfo : + // coordinatorServerInfos) { + // + // serverNodeList.add(coordinatorServerInfo.getNode()); + // } + // } else { + // // Fallback to single coordinator for backward + // compatibility + // ServerNode singleCoordinator = + // cluster.getCoordinatorServer(); + // if (singleCoordinator != null) { + // serverNodeList.add(singleCoordinator); + // } + // } serverNodeList.add(cluster.getCoordinatorServer()); serverNodeList.addAll(cluster.getAliveTabletServerList()); future.complete(serverNodeList); @@ -209,6 +231,33 @@ public CompletableFuture> getServerNodes() { return future; } + @Override + public CompletableFuture> describeCoordinators() { + CompletableFuture> future = new CompletableFuture<>(); + CompletableFuture.runAsync( + () -> { + try { + Cluster cluster = + sendMetadataRequestAndRebuildCluster( + readOnlyGateway, + false, + metadataUpdater.getCluster(), + null, + null, + null); + List coordinatorServerInfos = + cluster.getCoordinatorServerInfos(); + future.complete( + coordinatorServerInfos != null + ? coordinatorServerInfos + : new ArrayList<>()); + } catch (Throwable t) { + future.completeExceptionally(t); + } + }); + return future; + } + @Override public CompletableFuture getTableSchema(TablePath tablePath) { GetTableSchemaRequest request = new GetTableSchemaRequest(); 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 ca291f6a36..1ab716902e 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 @@ -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 coordinatorServerInfos = + getCoordinatorServerInfos(response); Map newTablePathToTableId; Map> newBucketLocations; @@ -159,6 +164,7 @@ public static Cluster sendMetadataRequestAndRebuildCluster( return new Cluster( newAliveTabletServers, coordinatorServer, + coordinatorServerInfos, newBucketLocations, newTablePathToTableId, newPartitionIdByPath, @@ -299,6 +305,33 @@ private static ServerNode getCoordinatorServer(MetadataResponse response) { } } + @Nullable + private static List getCoordinatorServerInfos( + MetadataResponse response) { + List pbCoordinatorServerInfos = + response.getCoordinatorServersList(); + if (pbCoordinatorServerInfos.isEmpty()) { + return null; + } + + List 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 getAliveTabletServers(MetadataResponse response) { Map aliveTabletServers = new HashMap<>(); response.getTabletServersList() diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java b/fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java index 41752f7dbc..f7932bfbdf 100644 --- a/fluss-common/src/main/java/org/apache/fluss/cluster/Cluster.java +++ b/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 coordinatorServerInfos; private final Map> availableLocationsByPath; private final Map availableLocationByBucket; private final Map aliveTabletServersById; @@ -66,7 +67,29 @@ public Cluster( Map tableIdByPath, Map partitionsIdByPath, Map bucketCountByTableOrPartition) { + this( + aliveTabletServersById, + coordinatorServer, + null, + bucketLocationsByPath, + tableIdByPath, + partitionsIdByPath, + bucketCountByTableOrPartition); + } + + public Cluster( + Map aliveTabletServersById, + @Nullable ServerNode coordinatorServer, + @Nullable List coordinatorServerInfos, + Map> bucketLocationsByPath, + Map tableIdByPath, + Map partitionsIdByPath, + Map 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 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 getCoordinatorServerInfos() { + return coordinatorServerInfos; + } + /** * @return The known set of alive tablet servers. */ diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorRole.java b/fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorRole.java new file mode 100644 index 0000000000..608e87b5b6 --- /dev/null +++ b/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; + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorServerInfo.java b/fluss-common/src/main/java/org/apache/fluss/cluster/CoordinatorServerInfo.java new file mode 100644 index 0000000000..d1bdb1eff4 --- /dev/null +++ b/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 + + '}'; + } +} 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 5e05819468..95ab83eb1e 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 @@ -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; @@ -88,6 +89,11 @@ public CompletableFuture> getServerNodes() { throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); } + @Override + public CompletableFuture> describeCoordinators() { + throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); + } + @Override public CompletableFuture getTableSchema(TablePath tablePath) { throw new UnsupportedOperationException("Not implemented in TestAdminAdapter"); diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index b778db8c24..3141c0cbfd 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -203,11 +203,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. diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java index 782c0629c2..6ce69d0fcd 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java @@ -18,6 +18,9 @@ package org.apache.fluss.server; import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.cluster.CoordinatorRole; +import org.apache.fluss.cluster.CoordinatorServerInfo; +import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; import org.apache.fluss.config.cluster.ConfigEntry; @@ -96,6 +99,7 @@ import org.apache.fluss.server.utils.ServerRpcMessageUtils; import org.apache.fluss.server.zk.ZooKeeperClient; import org.apache.fluss.server.zk.data.BucketSnapshot; +import org.apache.fluss.server.zk.data.CoordinatorAddress; import org.apache.fluss.server.zk.data.PartitionRegistration; import org.apache.fluss.server.zk.data.lake.LakeTableSnapshot; @@ -106,6 +110,7 @@ import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -740,8 +745,80 @@ protected MetadataResponse processMetadataRequest( ServerNode coordinatorServer = metadataCache.getCoordinatorServer(listenerName); Set aliveTabletServers = new HashSet<>(metadataCache.getAllAliveTabletServers(listenerName).values()); + + // Collect all coordinator servers with role and liveness information + List coordinatorServerInfos = new ArrayList<>(); + try { + // Get all registered coordinators from ZK + List allCoordinators = zkClient.getCoordinatorServers(); + Collections.sort( + allCoordinators, + (left, right) -> { + Endpoint leftEndpoint = getCoordinatorEndpoint(left, listenerName); + Endpoint rightEndpoint = getCoordinatorEndpoint(right, listenerName); + String leftHost = leftEndpoint == null ? "" : leftEndpoint.getHost(); + String rightHost = rightEndpoint == null ? "" : rightEndpoint.getHost(); + int hostCompare = leftHost.compareTo(rightHost); + if (hostCompare != 0) { + return hostCompare; + } + int leftPort = leftEndpoint == null ? -1 : leftEndpoint.getPort(); + int rightPort = rightEndpoint == null ? -1 : rightEndpoint.getPort(); + int portCompare = Integer.compare(leftPort, rightPort); + if (portCompare != 0) { + return portCompare; + } + return left.getId().compareTo(right.getId()); + }); + Set aliveCoordinatorIds = zkClient.getAliveCoordinatorServerIds(); + Optional leaderCoordinator = zkClient.getCoordinatorLeaderAddress(); + + for (int i = 0; i < allCoordinators.size(); i++) { + CoordinatorAddress coordinator = allCoordinators.get(i); + boolean isAlive = aliveCoordinatorIds.contains(coordinator.getId()); + CoordinatorRole role = + leaderCoordinator.isPresent() + && coordinator + .getId() + .equals(leaderCoordinator.get().getId()) + ? CoordinatorRole.LEADER + : CoordinatorRole.STANDBY; + Endpoint endpoint = getCoordinatorEndpoint(coordinator, listenerName); + if (endpoint != null) { + ServerNode node = + new ServerNode( + i, + endpoint.getHost(), + endpoint.getPort(), + ServerType.COORDINATOR); + coordinatorServerInfos.add( + new CoordinatorServerInfo(coordinator.getId(), node, role, isAlive)); + } + } + } catch (Exception e) { + LOG.warn("Failed to collect all coordinator servers from ZK", e); + } + return buildMetadataResponse( - coordinatorServer, aliveTabletServers, tablesMetadata, partitionsMetadata); + coordinatorServer, + coordinatorServerInfos, + aliveTabletServers, + tablesMetadata, + partitionsMetadata); + } + + @Nullable + private static Endpoint getCoordinatorEndpoint( + CoordinatorAddress coordinator, String listenerName) { + for (Endpoint endpoint : coordinator.getEndpoints()) { + if (listenerName.equals(endpoint.getListenerName())) { + return endpoint; + } + } + if (!coordinator.getEndpoints().isEmpty()) { + return coordinator.getEndpoints().get(0); + } + return null; } private boolean isPartitionAssignmentMissingFromZk(long partitionId) { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java index 242f92a0ea..e616161377 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java @@ -17,6 +17,7 @@ package org.apache.fluss.server.utils; +import org.apache.fluss.cluster.CoordinatorServerInfo; import org.apache.fluss.cluster.Endpoint; import org.apache.fluss.cluster.ServerNode; import org.apache.fluss.cluster.ServerType; @@ -108,6 +109,7 @@ import org.apache.fluss.rpc.messages.PbAlterConfig; import org.apache.fluss.rpc.messages.PbBucketMetadata; import org.apache.fluss.rpc.messages.PbBucketOffset; +import org.apache.fluss.rpc.messages.PbCoordinatorServerInfo; import org.apache.fluss.rpc.messages.PbCreateAclRespInfo; import org.apache.fluss.rpc.messages.PbDatabaseSummary; import org.apache.fluss.rpc.messages.PbDescribeConfig; @@ -461,6 +463,7 @@ private static TableChange.ColumnPosition toColumnPosition(int columnPositionTyp public static MetadataResponse buildMetadataResponse( @Nullable ServerNode coordinatorServer, + List coordinatorServerInfos, Set aliveTabletServers, List tableMetadataList, List partitionMetadataList) { @@ -474,6 +477,24 @@ public static MetadataResponse buildMetadataResponse( .setPort(coordinatorServer.port()); } + // Add all coordinator servers with role and liveness information + List pbCoordinatorServerInfos = new ArrayList<>(); + for (CoordinatorServerInfo coordinatorServerInfo : coordinatorServerInfos) { + ServerNode node = coordinatorServerInfo.getNode(); + int numericId = node.id(); + PbCoordinatorServerInfo pbCoordinatorServerInfo = + new PbCoordinatorServerInfo() + .setId(numericId) + .setRole(coordinatorServerInfo.getRole().getValue()) + .setIsAlive(coordinatorServerInfo.isAlive()); + pbCoordinatorServerInfo + .setCoordinatorServer() + .setNodeId(numericId) + .setHost(node.host()) + .setPort(node.port()); + pbCoordinatorServerInfos.add(pbCoordinatorServerInfo); + } + List pbServerNodeList = new ArrayList<>(); for (ServerNode serverNode : aliveTabletServers) { PbServerNode pbServerNode = @@ -496,12 +517,27 @@ public static MetadataResponse buildMetadataResponse( partitionMetadata -> pbPartitionMetadataList.add(toPbPartitionMetadata(partitionMetadata))); + metadataResponse.addAllCoordinatorServers(pbCoordinatorServerInfos); metadataResponse.addAllTabletServers(pbServerNodeList); metadataResponse.addAllTableMetadatas(pbTableMetadataList); metadataResponse.addAllPartitionMetadatas(pbPartitionMetadataList); return metadataResponse; } + // Overload for backward compatibility + public static MetadataResponse buildMetadataResponse( + @Nullable ServerNode coordinatorServer, + Set aliveTabletServers, + List tableMetadataList, + List partitionMetadataList) { + return buildMetadataResponse( + coordinatorServer, + Collections.emptyList(), + aliveTabletServers, + tableMetadataList, + partitionMetadataList); + } + public static UpdateMetadataRequest makeUpdateMetadataRequest( @Nullable ServerInfo coordinatorServer, @Nullable Integer coordinatorEpoch, diff --git a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java index 010b2e840e..1e7e83ef8e 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java @@ -110,6 +110,7 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -252,6 +253,13 @@ public void registerCoordinatorServer(CoordinatorAddress coordinatorAddress) thr LOG.info("Registered Coordinator server {} at path {}.", coordinatorAddress, path); } + /** Get the registered coordinator server information for the given coordinator id. */ + public Optional getCoordinatorServer(String coordinatorId) + throws Exception { + Optional bytes = getOrEmpty(ZkData.CoordinatorIdZNode.path(coordinatorId)); + return bytes.map(ZkData.CoordinatorIdZNode::decode); + } + /** * Become coordinator leader. This method is a step after electCoordinatorLeader() and before * registerCoordinatorLeader(). This is to ensure the coordinator get and update the coordinator @@ -336,6 +344,38 @@ public List getCoordinatorServerList() throws Exception { return getChildren(ZkData.CoordinatorIdsZNode.path()); } + /** + * Gets the registered coordinator server information for all coordinator ids stored in + * ZooKeeper. + */ + public List getCoordinatorServers() throws Exception { + List coordinatorServers = new ArrayList<>(); + for (String coordinatorId : getCoordinatorServerList()) { + Optional coordinatorServer = getCoordinatorServer(coordinatorId); + coordinatorServer.ifPresent(coordinatorServers::add); + } + return coordinatorServers; + } + + /** Gets the coordinator ids currently participating in the election group. */ + public List getAliveCoordinatorServerList() throws Exception { + List aliveCoordinatorServerIds = new ArrayList<>(); + for (String childName : getChildren(ZkData.CoordinatorElectionZNode.path())) { + String childPath = ZkData.CoordinatorElectionZNode.path() + "/" + childName; + Optional bytes = getOrEmpty(childPath); + if (bytes.isPresent() && bytes.get().length > 0) { + aliveCoordinatorServerIds.add(new String(bytes.get(), StandardCharsets.UTF_8)); + } + } + Collections.sort(aliveCoordinatorServerIds); + return aliveCoordinatorServerIds; + } + + /** Gets the ids of all coordinator servers that are currently alive. */ + public Set getAliveCoordinatorServerIds() throws Exception { + return new HashSet<>(getAliveCoordinatorServerList()); + } + /** Ensure epoch znode exists. */ public void ensureEpochZnodeExists() throws Exception { String path = ZkData.CoordinatorEpochZNode.path(); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java index 60020824a5..9176c2089d 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java @@ -43,9 +43,11 @@ import org.apache.fluss.server.zk.data.TableRegistration; import org.apache.fluss.server.zk.data.TabletServerRegistration; import org.apache.fluss.server.zk.data.ZkData.BucketIdZNode; +import org.apache.fluss.server.zk.data.ZkData.CoordinatorElectionZNode; import org.apache.fluss.server.zk.data.lease.KvSnapshotLeaseMetadata; import org.apache.fluss.shaded.curator5.org.apache.curator.CuratorZookeeperClient; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework; +import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.CreateMode; import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.KeeperException; import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.ZooKeeper; import org.apache.fluss.shaded.zookeeper3.org.apache.zookeeper.client.ZKClientConfig; @@ -62,6 +64,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -134,6 +137,46 @@ void testCoordinatorLeader() throws Exception { assertThat(gottenAddress).isEqualTo(coordinatorAddress); } + @Test + void testCoordinatorServerListAndAliveCoordinatorServerList() throws Exception { + CoordinatorAddress coordinatorAddress1 = + new CoordinatorAddress( + "coordinator-1", Endpoint.fromListenersString("CLIENT://localhost1:10012")); + CoordinatorAddress coordinatorAddress2 = + new CoordinatorAddress( + "coordinator-2", Endpoint.fromListenersString("CLIENT://localhost2:10013")); + + zookeeperClient.registerCoordinatorServer(coordinatorAddress1); + zookeeperClient.registerCoordinatorServer(coordinatorAddress2); + zookeeperClient.registerCoordinatorLeader(coordinatorAddress2); + + zookeeperClient + .getCuratorClient() + .create() + .creatingParentsIfNeeded() + .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) + .forPath( + CoordinatorElectionZNode.path() + "/latch-", + "coordinator-1".getBytes(StandardCharsets.UTF_8)); + zookeeperClient + .getCuratorClient() + .create() + .creatingParentsIfNeeded() + .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) + .forPath( + CoordinatorElectionZNode.path() + "/latch-", + "coordinator-2".getBytes(StandardCharsets.UTF_8)); + + assertThat(zookeeperClient.getCoordinatorServerList()) + .containsExactlyInAnyOrder("coordinator-1", "coordinator-2"); + assertThat(zookeeperClient.getCoordinatorServers()) + .containsExactlyInAnyOrder(coordinatorAddress1, coordinatorAddress2); + assertThat(zookeeperClient.getAliveCoordinatorServerList()) + .containsExactlyInAnyOrder("coordinator-1", "coordinator-2"); + assertThat(zookeeperClient.getAliveCoordinatorServerIds()) + .containsExactlyInAnyOrder("coordinator-1", "coordinator-2"); + } + @Test void testTabletServer() throws Exception { // try to get tablet server, should return empty From 9df0a31e39ca6889f5215fcd304cf11baf060009 Mon Sep 17 00:00:00 2001 From: MaheshS08 Date: Thu, 17 Sep 2026 07:44:01 +0530 Subject: [PATCH 2/3] [server] Remove commented-out code for coordinator server retrieval in FlussAdmin --- .../apache/fluss/client/admin/FlussAdmin.java | 21 ------------------- 1 file changed, 21 deletions(-) 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 5f5207b817..54c490a991 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 @@ -200,27 +200,6 @@ public CompletableFuture> getServerNodes() { null, null, null); - // Add all coordinator servers (leader + standbys) - // List coordinatorServerInfos - // = - // cluster.getCoordinatorServerInfos(); - // if (coordinatorServerInfos != null && - // !coordinatorServerInfos.isEmpty()) { - // for (CoordinatorServerInfo - // coordinatorServerInfo : - // coordinatorServerInfos) { - // - // serverNodeList.add(coordinatorServerInfo.getNode()); - // } - // } else { - // // Fallback to single coordinator for backward - // compatibility - // ServerNode singleCoordinator = - // cluster.getCoordinatorServer(); - // if (singleCoordinator != null) { - // serverNodeList.add(singleCoordinator); - // } - // } serverNodeList.add(cluster.getCoordinatorServer()); serverNodeList.addAll(cluster.getAliveTabletServerList()); future.complete(serverNodeList); From 8726bdf6341c95665e15023fcf33e3e526d84532 Mon Sep 17 00:00:00 2001 From: MaheshS08 Date: Sun, 20 Sep 2026 20:36:18 +0530 Subject: [PATCH 3/3] CI/CD fixes --- .../src/main/java/org/apache/fluss/server/RpcServiceBase.java | 1 + .../java/org/apache/fluss/server/zk/ZooKeeperClientTest.java | 1 + 2 files changed, 2 insertions(+) diff --git a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java index 0be1fae10e..fc7bcacc4b 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java @@ -115,6 +115,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.Comparator; import java.util.HashSet; import java.util.List; import java.util.Map; diff --git a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java index 6dc6bc924f..6f22945d15 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/zk/ZooKeeperClientTest.java @@ -47,6 +47,7 @@ import org.apache.fluss.server.zk.data.TabletServerRegistration; import org.apache.fluss.server.zk.data.ZkData.BucketIdZNode; import org.apache.fluss.server.zk.data.ZkData.CoordinatorElectionZNode; +import org.apache.fluss.server.zk.data.ZkData.TableIdZNode; import org.apache.fluss.server.zk.data.lease.KvSnapshotLeaseMetadata; import org.apache.fluss.shaded.curator5.org.apache.curator.CuratorZookeeperClient; import org.apache.fluss.shaded.curator5.org.apache.curator.framework.CuratorFramework;