[client] Add retry when get one available tablet serverNode fails (#425)

Author: zhaozijun109

fluss-client/src/main/java/com/alibaba/fluss/client/FlussConnection.java

@@ -139,14 +139,14 @@ public RemoteFileDownloader getOrCreateRemoteFileDownloader() {
                 if (securityTokenManager == null) {
                     // prepare security token manager
                     // create the admin read only gateway
-                    // todo: may add retry logic when no any available tablet server?
                     AdminReadOnlyGateway gateway =
                             GatewayClientProxy.createGatewayProxy(
                                     () ->
                                             getOneAvailableTabletServerNode(
                                                     metadataUpdater.getCluster()),
                                     rpcClient,
                                     AdminReadOnlyGateway.class);
+
                     SecurityTokenProvider securityTokenProvider =
                             new DefaultSecurityTokenProvider(gateway);
                     securityTokenManager =

fluss-client/src/main/java/com/alibaba/fluss/client/utils/MetadataUtils.java

@@ -37,6 +37,9 @@
 import com.alibaba.fluss.rpc.messages.PbTableMetadata;
 import com.alibaba.fluss.rpc.messages.PbTablePath;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -52,6 +55,9 @@
 
 /** Utils for metadata for client. */
 public class MetadataUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(MetadataUtils.class);
+
+    private static final int MAX_RETRY_TIMES = 5;
 
     private static final Random randOffset = new Random();
 
@@ -255,10 +261,30 @@ public NewTableMetadata(
     }
 
     public static ServerNode getOneAvailableTabletServerNode(Cluster cluster) {
-        List<ServerNode> aliveTabletServers = cluster.getAliveTabletServerList();
-        if (aliveTabletServers.isEmpty()) {
-            throw new FlussRuntimeException("no alive tablet server in cluster");
+        List<ServerNode> aliveTabletServers = null;
+        for (int retryTimes = 0; retryTimes <= MAX_RETRY_TIMES; retryTimes++) {
+            aliveTabletServers = cluster.getAliveTabletServerList();
+            if (aliveTabletServers.isEmpty()) {
+                LOG.error("Fluss create gateway proxy error, retry times = {}.", retryTimes);
+                if (retryTimes >= MAX_RETRY_TIMES) {
+                    String exceptionMsg =
+                            String.format(
+                                    "Execution of Fluss get one available tablet failed, no alive tablet server in cluster, retry times = %d.",
+                                    retryTimes);
+                    throw new FlussRuntimeException(exceptionMsg);
+                } else {
+                    try {
+                        Thread.sleep(1000L * retryTimes);
+                    } catch (InterruptedException interruptedException) {
+                        Thread.currentThread().interrupt();
+                        throw new RuntimeException(interruptedException);
+                    }
+                }
+            } else {
+                break;
+            }
         }
+
         // just pick one random server node
         int offset = randOffset.nextInt(aliveTabletServers.size());
         return aliveTabletServers.get(offset);

fluss-client/src/test/java/com/alibaba/fluss/client/table/FlussFailServerTableITCase.java

@@ -16,12 +16,17 @@
 
 package com.alibaba.fluss.client.table;
 
+import com.alibaba.fluss.client.Connection;
+import com.alibaba.fluss.client.ConnectionFactory;
 import com.alibaba.fluss.client.admin.ClientToServerITCaseBase;
 import com.alibaba.fluss.client.table.scanner.ScanRecord;
 import com.alibaba.fluss.client.table.scanner.log.LogScanner;
 import com.alibaba.fluss.client.table.scanner.log.ScanRecords;
 import com.alibaba.fluss.client.table.writer.AppendWriter;
 import com.alibaba.fluss.client.table.writer.UpsertWriter;
+import com.alibaba.fluss.cluster.ServerNode;
+import com.alibaba.fluss.cluster.ServerType;
+import com.alibaba.fluss.exception.FlussRuntimeException;
 import com.alibaba.fluss.row.GenericRow;
 import com.alibaba.fluss.row.InternalRow;
 
@@ -31,6 +36,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import static com.alibaba.fluss.record.TestData.DATA1_ROW_TYPE;
 import static com.alibaba.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR;
@@ -39,6 +45,7 @@
 import static com.alibaba.fluss.record.TestData.DATA1_TABLE_PATH_PK;
 import static com.alibaba.fluss.testutils.DataTestUtils.row;
 import static com.alibaba.fluss.testutils.InternalRowListAssert.assertThatRows;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /** IT case for {@link FlussTable} in the case of one tablet server fails. */
 class FlussFailServerTableITCase extends ClientToServerITCaseBase {
@@ -146,6 +153,39 @@ void testLogScan() throws Exception {
         }
     }
 
+    @Test
+    void testRetryGetTabletServerNodes() throws Exception {
+        createTable(DATA1_TABLE_PATH, DATA1_TABLE_DESCRIPTOR, false);
+        try (Table table = conn.getTable(DATA1_TABLE_PATH)) {
+            table.newScan().createLogScanner();
+
+            List<ServerNode> serverNodes =
+                    conn.getAdmin().getServerNodes().get().stream()
+                            .filter(
+                                    serverNode ->
+                                            serverNode.serverType() == ServerType.TABLET_SERVER)
+                            .collect(Collectors.toList());
+
+            // kill all tablet server
+            for (ServerNode serverNode : serverNodes) {
+                FLUSS_CLUSTER_EXTENSION.stopTabletServer(serverNode.id());
+            }
+
+            try (Connection connNew = ConnectionFactory.createConnection(clientConf)) {
+                assertThatThrownBy(() -> connNew.getTable(DATA1_TABLE_PATH))
+                        .cause()
+                        .isInstanceOf(FlussRuntimeException.class)
+                        .hasMessage(
+                                "Execution of Fluss get one available tablet failed, no alive tablet server in cluster, retry times = %d.",
+                                5);
+            } finally {
+                for (ServerNode serverNode : serverNodes) {
+                    FLUSS_CLUSTER_EXTENSION.startTabletServer(serverNode.id());
+                }
+            }
+        }
+    }
+
     private List<InternalRow> toRows(ScanRecords scanRecords) {
         List<InternalRow> rows = new ArrayList<>();
         for (ScanRecord scanRecord : scanRecords) {