[client] Support initial cluster with retries to avoid initial failed when bootstrap server only contains one available address (#2024)

Author: swuferhong

fluss-client/src/main/java/org/apache/fluss/client/metadata/MetadataUpdater.java

@@ -26,8 +26,10 @@
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.exception.FlussRuntimeException;
+import org.apache.fluss.exception.NetworkException;
 import org.apache.fluss.exception.PartitionNotExistException;
 import org.apache.fluss.exception.RetriableException;
+import org.apache.fluss.exception.StaleMetadataException;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
@@ -38,7 +40,6 @@
 import org.apache.fluss.rpc.gateway.AdminReadOnlyGateway;
 import org.apache.fluss.rpc.gateway.CoordinatorGateway;
 import org.apache.fluss.rpc.gateway.TabletServerGateway;
-import org.apache.fluss.utils.ExceptionUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -56,12 +57,14 @@
 import java.util.stream.Collectors;
 
 import static org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster;
+import static org.apache.fluss.utils.ExceptionUtils.stripExecutionException;
 
 /** The updater to initialize and update client metadata. */
 public class MetadataUpdater {
     private static final Logger LOG = LoggerFactory.getLogger(MetadataUpdater.class);
 
-    private static final int MAX_RETRY_TIMES = 5;
+    private static final int MAX_RETRY_TIMES = 3;
+    private static final int RETRY_INTERVAL_MS = 100;
 
     private final RpcClient rpcClient;
     protected volatile Cluster cluster;
@@ -270,7 +273,7 @@ public void updateMetadata(
                                 tablePartitionIds);
             }
         } catch (Exception e) {
-            Throwable t = ExceptionUtils.stripExecutionException(e);
+            Throwable t = stripExecutionException(e);
             if (t instanceof RetriableException || t instanceof TimeoutException) {
                 LOG.warn("Failed to update metadata, but the exception is re-triable.", t);
             } else if (t instanceof PartitionNotExistException) {
@@ -292,10 +295,33 @@ private static Cluster initializeCluster(Configuration conf, RpcClient rpcClient
         Cluster cluster = null;
         Exception lastException = null;
         for (InetSocketAddress address : inetSocketAddresses) {
+            ServerNode serverNode = null;
             try {
-                cluster = tryToInitializeCluster(rpcClient, address);
-                break;
+                serverNode =
+                        new ServerNode(
+                                -1,
+                                address.getHostString(),
+                                address.getPort(),
+                                ServerType.COORDINATOR);
+                ServerNode finalServerNode = serverNode;
+                AdminReadOnlyGateway adminReadOnlyGateway =
+                        GatewayClientProxy.createGatewayProxy(
+                                () -> finalServerNode, rpcClient, AdminReadOnlyGateway.class);
+                if (inetSocketAddresses.size() == 1) {
+                    // if there is only one bootstrap server, we can retry to connect to it.
+                    cluster =
+                            tryToInitializeClusterWithRetries(
+                                    rpcClient, serverNode, adminReadOnlyGateway, MAX_RETRY_TIMES);
+                } else {
+                    cluster = tryToInitializeCluster(adminReadOnlyGateway);
+                    break;
+                }
             } catch (Exception e) {
+                // We should dis-connected with the bootstrap server id to make sure the next
+                // retry can rebuild the connection.
+                if (serverNode != null) {
+                    rpcClient.disconnect(serverNode.uid());
+                }
                 LOG.error(
                         "Failed to initialize fluss client connection to bootstrap server: {}",
                         address,
@@ -306,24 +332,64 @@ private static Cluster initializeCluster(Configuration conf, RpcClient rpcClient
 
         if (cluster == null && lastException != null) {
             String errorMsg =
-                    "Failed to initialize fluss client connection to server because no "
-                            + "bootstrap server is validate. bootstrap servers: "
-                            + inetSocketAddresses;
+                    "Failed to initialize fluss client connection to bootstrap servers: "
+                            + inetSocketAddresses
+                            + ". \nReason: "
+                            + lastException.getMessage();
             LOG.error(errorMsg);
             throw new IllegalStateException(errorMsg, lastException);
         }
 
         return cluster;
     }
 
-    private static Cluster tryToInitializeCluster(RpcClient rpcClient, InetSocketAddress address)
+    @VisibleForTesting
+    static @Nullable Cluster tryToInitializeClusterWithRetries(
+            RpcClient rpcClient,
+            ServerNode serverNode,
+            AdminReadOnlyGateway gateway,
+            int maxRetryTimes)
+            throws Exception {
+        int retryCount = 0;
+        while (retryCount <= maxRetryTimes) {
+            try {
+                return tryToInitializeCluster(gateway);
+            } catch (Exception e) {
+                Throwable cause = stripExecutionException(e);
+                // in case of bootstrap is recovering, we should retry to connect.
+                if (!(cause instanceof StaleMetadataException || cause instanceof NetworkException)
+                        || retryCount >= maxRetryTimes) {
+                    throw e;
+                }
+
+                // We should dis-connected with the bootstrap server id to make sure the next
+                // retry can rebuild the connection.
+                rpcClient.disconnect(serverNode.uid());
+
+                long delayMs = (long) (RETRY_INTERVAL_MS * Math.pow(2, retryCount));
+                LOG.warn(
+                        "Failed to connect to bootstrap server: {} (retry {}/{}). Retrying in {} ms.",
+                        serverNode,
+                        retryCount + 1,
+                        maxRetryTimes,
+                        delayMs,
+                        e);
+
+                try {
+                    Thread.sleep(delayMs);
+                } catch (InterruptedException ex) {
+                    Thread.currentThread().interrupt();
+                    throw new RuntimeException("Interrupted during retry sleep", ex);
+                }
+                retryCount++;
+            }
+        }
+
+        return null;
+    }
+
+    private static Cluster tryToInitializeCluster(AdminReadOnlyGateway adminReadOnlyGateway)
             throws Exception {
-        ServerNode serverNode =
-                new ServerNode(
-                        -1, address.getHostString(), address.getPort(), ServerType.COORDINATOR);
-        AdminReadOnlyGateway adminReadOnlyGateway =
-                GatewayClientProxy.createGatewayProxy(
-                        () -> serverNode, rpcClient, AdminReadOnlyGateway.class);
         return sendMetadataRequestAndRebuildCluster(adminReadOnlyGateway, Collections.emptySet());
     }
 

fluss-client/src/test/java/org/apache/fluss/client/metadata/MetadataUpdaterITCase.java

@@ -0,0 +1,124 @@
+/*
+ * 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.metadata;
+
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.cluster.Cluster;
+import org.apache.fluss.cluster.ServerNode;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.rpc.RpcClient;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT test for update metadata of {@link MetadataUpdater}. */
+class MetadataUpdaterITCase {
+
+    @RegisterExtension
+    public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+            FlussClusterExtension.builder().setNumOfTabletServers(2).build();
+
+    @Test
+    void testRebuildClusterNTimes() throws Exception {
+        Configuration clientConf = FLUSS_CLUSTER_EXTENSION.getClientConfig();
+        Connection conn = ConnectionFactory.createConnection(clientConf);
+        Admin admin = conn.getAdmin();
+        TablePath tablePath = TablePath.of("fluss", "test");
+        admin.createTable(tablePath, DATA1_TABLE_DESCRIPTOR, true).get();
+        admin.close();
+        conn.close();
+
+        RpcClient rpcClient = FLUSS_CLUSTER_EXTENSION.getRpcClient();
+        MetadataUpdater metadataUpdater = new MetadataUpdater(clientConf, rpcClient);
+        // update metadata
+        metadataUpdater.updateMetadata(Collections.singleton(tablePath), null, null);
+        Cluster cluster = metadataUpdater.getCluster();
+
+        // repeat 20K times to reproduce StackOverflowError if there is
+        // any N levels UnmodifiableCollection
+        for (int i = 0; i < 20000; i++) {
+            cluster =
+                    sendMetadataRequestAndRebuildCluster(
+                            FLUSS_CLUSTER_EXTENSION.newCoordinatorClient(),
+                            true,
+                            cluster,
+                            null,
+                            null,
+                            null);
+        }
+    }
+
+    @Test
+    void testUpdateWithEmptyMetadataResponse() throws Exception {
+        RpcClient rpcClient = FLUSS_CLUSTER_EXTENSION.getRpcClient();
+        MetadataUpdater metadataUpdater =
+                new MetadataUpdater(FLUSS_CLUSTER_EXTENSION.getClientConfig(), rpcClient);
+
+        // update metadata
+        metadataUpdater.updateMetadata(null, null, null);
+        Cluster cluster = metadataUpdater.getCluster();
+
+        List<ServerNode> expectedServerNodes = FLUSS_CLUSTER_EXTENSION.getTabletServerNodes();
+        assertThat(expectedServerNodes).hasSize(2);
+        assertThat(cluster.getAliveTabletServerList()).isEqualTo(expectedServerNodes);
+
+        // then, stop coordinator server, can still update metadata
+        FLUSS_CLUSTER_EXTENSION.stopCoordinatorServer();
+        metadataUpdater.updateMetadata(null, null, null);
+        assertThat(cluster.getAliveTabletServerList()).isEqualTo(expectedServerNodes);
+
+        // start a new tablet server, the tablet server will return empty metadata
+        // response since no coordinator server to send newest metadata to the tablet server
+        int newServerId = 2;
+        FLUSS_CLUSTER_EXTENSION.startTabletServer(newServerId);
+
+        // we mock a new cluster with only server 1 so that it'll only send request
+        // to server 1, which will return empty resonate
+        Cluster newCluster =
+                new Cluster(
+                        Collections.singletonMap(
+                                newServerId,
+                                FLUSS_CLUSTER_EXTENSION.getTabletServerNodes().get(newServerId)),
+                        null,
+                        Collections.emptyMap(),
+                        Collections.emptyMap(),
+                        Collections.emptyMap(),
+                        Collections.emptyMap());
+
+        metadataUpdater = new MetadataUpdater(rpcClient, newCluster);
+        // shouldn't update metadata to empty since the empty metadata will be ignored
+        metadataUpdater.updateMetadata(null, null, null);
+        assertThat(metadataUpdater.getCluster().getAliveTabletServers())
+                .isEqualTo(newCluster.getAliveTabletServers())
+                .hasSize(1);
+
+        // recover the coordinator
+        FLUSS_CLUSTER_EXTENSION.startCoordinatorServer();
+    }
+}