[client] Add unavailable tabletServer set for MetadataUpdater to avoid serverNode info of all tabletServers are invalid (#2098)

swuferhong · web-flow · commit 61c4ece67ef9 · 2025-12-04T13:50:08.000+08:00
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java b/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java
@@ -29,6 +29,7 @@
 import org.apache.fluss.client.token.SecurityTokenManager;
 import org.apache.fluss.client.token.SecurityTokenProvider;
 import org.apache.fluss.client.write.WriterClient;
+import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
@@ -42,6 +43,7 @@
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 
 import static org.apache.fluss.client.utils.MetadataUtils.getOneAvailableTabletServerNode;
@@ -162,9 +164,17 @@ public RemoteFileDownloader getOrCreateRemoteFileDownloader() {
                     // todo: may add retry logic when no any available tablet server?
                     AdminReadOnlyGateway gateway =
                             GatewayClientProxy.createGatewayProxy(
-                                    () ->
-                                            getOneAvailableTabletServerNode(
-                                                    metadataUpdater.getCluster()),
+                                    () -> {
+                                        ServerNode serverNode =
+                                                getOneAvailableTabletServerNode(
+                                                        metadataUpdater.getCluster(),
+                                                        new HashSet<>());
+                                        if (serverNode == null) {
+                                            throw new FlussRuntimeException(
+                                                    "no available tablet server");
+                                        }
+                                        return serverNode;
+                                    },
                                     rpcClient,
                                     AdminReadOnlyGateway.class);
                     SecurityTokenProvider securityTokenProvider =
diff --git a/fluss-client/src/main/java/org/apache/fluss/client/metadata/MetadataUpdater.java b/fluss-client/src/main/java/org/apache/fluss/client/metadata/MetadataUpdater.java
@@ -50,11 +50,14 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 
+import static org.apache.fluss.client.utils.MetadataUtils.getOneAvailableTabletServerNode;
 import static org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster;
 import static org.apache.fluss.utils.ExceptionUtils.stripExecutionException;
 
@@ -65,16 +68,19 @@ public class MetadataUpdater {
     private static final int MAX_RETRY_TIMES = 3;
     private static final int RETRY_INTERVAL_MS = 100;
 
+    private final Configuration conf;
     private final RpcClient rpcClient;
+    private final Set<Integer> unavailableTabletServerIds = new CopyOnWriteArraySet<>();
     protected volatile Cluster cluster;
 
-    public MetadataUpdater(Configuration configuration, RpcClient rpcClient) {
-        this(rpcClient, initializeCluster(configuration, rpcClient));
+    public MetadataUpdater(Configuration conf, RpcClient rpcClient) {
+        this(rpcClient, conf, initializeCluster(conf, rpcClient));
     }
 
     @VisibleForTesting
-    public MetadataUpdater(RpcClient rpcClient, Cluster cluster) {
+    public MetadataUpdater(RpcClient rpcClient, Configuration conf, Cluster cluster) {
         this.rpcClient = rpcClient;
+        this.conf = conf;
         this.cluster = cluster;
     }
 
@@ -240,19 +246,43 @@ public void updateMetadata(
             @Nullable Collection<PhysicalTablePath> tablePartitionNames,
             @Nullable Collection<Long> tablePartitionIds)
             throws PartitionNotExistException {
+        ServerNode serverNode =
+                getOneAvailableTabletServerNode(cluster, unavailableTabletServerIds);
         try {
             synchronized (this) {
-                cluster =
-                        sendMetadataRequestAndRebuildCluster(
-                                cluster,
-                                rpcClient,
-                                tablePaths,
-                                tablePartitionNames,
-                                tablePartitionIds);
+                if (serverNode == null) {
+                    LOG.info(
+                            "No available tablet server to update metadata, try to re-initialize cluster using bootstrap server.");
+                    cluster = initializeCluster(conf, rpcClient);
+                } else {
+                    cluster =
+                            sendMetadataRequestAndRebuildCluster(
+                                    cluster,
+                                    rpcClient,
+                                    tablePaths,
+                                    tablePartitionNames,
+                                    tablePartitionIds,
+                                    serverNode);
+                }
+            }
+
+            Map<Integer, ServerNode> aliveTabletServers = cluster.getAliveTabletServers();
+            unavailableTabletServerIds.removeIf(aliveTabletServers::containsKey);
+            if (!unavailableTabletServerIds.isEmpty()) {
+                LOG.info(
+                        "After update metadata, unavailable tabletServer set: {}",
+                        unavailableTabletServerIds);
             }
         } catch (Exception e) {
             Throwable t = stripExecutionException(e);
             if (t instanceof RetriableException || t instanceof TimeoutException) {
+                if (serverNode != null) {
+                    unavailableTabletServerIds.add(serverNode.id());
+                    LOG.warn(
+                            "tabletServer {} is unavailable for updating metadata for retriable exception. unavailable tabletServer set {}",
+                            serverNode,
+                            unavailableTabletServerIds);
+                }
                 LOG.warn("Failed to update metadata, but the exception is re-triable.", t);
             } else if (t instanceof PartitionNotExistException) {
                 LOG.warn("Failed to update metadata because the partition does not exist", t);
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
@@ -21,7 +21,6 @@
 import org.apache.fluss.cluster.Cluster;
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.cluster.ServerType;
-import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.exception.StaleMetadataException;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
@@ -39,6 +38,9 @@
 import org.apache.fluss.rpc.messages.PbTableMetadata;
 import org.apache.fluss.rpc.messages.PbTablePath;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -55,6 +57,8 @@
 /** Utils for metadata for client. */
 public class MetadataUtils {
 
+    private static final Logger LOG = LoggerFactory.getLogger(MetadataUtils.class);
+
     private static final Random randOffset = new Random();
 
     /**
@@ -79,13 +83,12 @@ public static Cluster sendMetadataRequestAndRebuildCluster(
             RpcClient client,
             @Nullable Set<TablePath> tablePaths,
             @Nullable Collection<PhysicalTablePath> tablePartitionNames,
-            @Nullable Collection<Long> tablePartitionIds)
+            @Nullable Collection<Long> tablePartitionIds,
+            ServerNode serverNode)
             throws ExecutionException, InterruptedException, TimeoutException {
         AdminReadOnlyGateway gateway =
                 GatewayClientProxy.createGatewayProxy(
-                        () -> getOneAvailableTabletServerNode(cluster),
-                        client,
-                        AdminReadOnlyGateway.class);
+                        () -> serverNode, client, AdminReadOnlyGateway.class);
         return sendMetadataRequestAndRebuildCluster(
                 gateway, true, cluster, tablePaths, tablePartitionNames, tablePartitionIds);
     }
@@ -253,10 +256,16 @@ public NewTableMetadata(
         }
     }
 
-    public static ServerNode getOneAvailableTabletServerNode(Cluster cluster) {
-        List<ServerNode> aliveTabletServers = cluster.getAliveTabletServerList();
+    public static @Nullable ServerNode getOneAvailableTabletServerNode(
+            Cluster cluster, Set<Integer> unavailableTabletServerIds) {
+        List<ServerNode> aliveTabletServers = new ArrayList<>(cluster.getAliveTabletServerList());
+        if (!unavailableTabletServerIds.isEmpty()) {
+            aliveTabletServers.removeIf(
+                    serverNode -> unavailableTabletServerIds.contains(serverNode.id()));
+        }
+
         if (aliveTabletServers.isEmpty()) {
-            throw new FlussRuntimeException("no alive tablet server in cluster");
+            return null;
         }
         // just pick one random server node
         int offset = randOffset.nextInt(aliveTabletServers.size());
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/MetadataUpdaterITCase.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/MetadataUpdaterITCase.java
@@ -111,7 +111,7 @@ void testUpdateWithEmptyMetadataResponse() throws Exception {
                         Collections.emptyMap(),
                         Collections.emptyMap());
 
-        metadataUpdater = new MetadataUpdater(rpcClient, newCluster);
+        metadataUpdater = new MetadataUpdater(rpcClient, new Configuration(), newCluster);
         // shouldn't update metadata to empty since the empty metadata will be ignored
         metadataUpdater.updateMetadata(null, null, null);
         assertThat(metadataUpdater.getCluster().getAliveTabletServers())
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java b/fluss-client/src/test/java/org/apache/fluss/client/metadata/TestingMetadataUpdater.java
@@ -55,17 +55,23 @@ public class TestingMetadataUpdater extends MetadataUpdater {
     private final Map<Integer, TestTabletServerGateway> tabletServerGatewayMap;
 
     public TestingMetadataUpdater(Map<TablePath, TableInfo> tableInfos) {
-        this(COORDINATOR, Arrays.asList(NODE1, NODE2, NODE3), tableInfos, null);
+        this(
+                COORDINATOR,
+                Arrays.asList(NODE1, NODE2, NODE3),
+                tableInfos,
+                null,
+                new Configuration());
     }
 
     private TestingMetadataUpdater(
             ServerNode coordinatorServer,
             List<ServerNode> tabletServers,
             Map<TablePath, TableInfo> tableInfos,
-            Map<Integer, TestTabletServerGateway> customGateways) {
+            Map<Integer, TestTabletServerGateway> customGateways,
+            Configuration conf) {
         super(
-                RpcClient.create(
-                        new Configuration(), TestingClientMetricGroup.newInstance(), false),
+                RpcClient.create(conf, TestingClientMetricGroup.newInstance(), false),
+                conf,
                 Cluster.empty());
         initializeCluster(coordinatorServer, tabletServers, tableInfos);
         coordinatorGateway = new TestCoordinatorGateway();
@@ -122,7 +128,8 @@ public TestingMetadataUpdater build() {
                     COORDINATOR,
                     Arrays.asList(NODE1, NODE2, NODE3),
                     tableInfos,
-                    customGateways.isEmpty() ? null : customGateways);
+                    customGateways.isEmpty() ? null : customGateways,
+                    new Configuration());
         }
     }
 
diff --git a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherTest.java
@@ -262,7 +262,7 @@ void testFetchWhenDestinationIsNullInMetadata() throws Exception {
                         oldCluster.getTableIdByPath(),
                         oldCluster.getPartitionIdByPath(),
                         oldCluster.getTableInfoByPath());
-        metadataUpdater = new MetadataUpdater(rpcClient, newCluster);
+        metadataUpdater = new MetadataUpdater(rpcClient, clientConf, newCluster);
 
         LogScannerStatus logScannerStatus = new LogScannerStatus();
         logScannerStatus.assignScanBuckets(Collections.singletonMap(tb0, 0L));
