[test] Fix some unstable test cases while running CI

Author: swuferhong

fluss-client/src/main/java/com/alibaba/fluss/client/table/FlussTable.java

@@ -212,15 +212,17 @@ private int getBucketId(byte[] keyBytes, InternalRow key) {
     public CompletableFuture<List<ScanRecord>> limitScan(
             TableBucket tableBucket, int limit, @Nullable int[] projectedFields) {
         // because that rocksdb is not suitable to projection, thus do it in client.
-        int leader = metadataUpdater.leaderFor(tableBucket);
         LimitScanRequest limitScanRequest =
                 new LimitScanRequest()
                         .setTableId(tableBucket.getTableId())
                         .setBucketId(tableBucket.getBucket())
                         .setLimit(limit);
         if (tableBucket.getPartitionId() != null) {
             limitScanRequest.setPartitionId(tableBucket.getPartitionId());
+            metadataUpdater.checkAndUpdateMetadata(tablePath, tableBucket);
         }
+
+        int leader = metadataUpdater.leaderFor(tableBucket);
         TabletServerGateway gateway = metadataUpdater.newTabletServerClientForNode(leader);
 
         CompletableFuture<List<ScanRecord>> future = new CompletableFuture<>();

…ient/admin/ClientToServerITCaseBase.java …ent/admin/ClientToServerITCaseUtils.javafluss-client/src/test/java/com/alibaba/fluss/client/admin/ClientToServerITCaseBase.java renamed to fluss-client/src/test/java/com/alibaba/fluss/client/admin/ClientToServerITCaseUtils.java fluss-client/src/test/java/com/alibaba/fluss/client/admin/ClientToServerITCaseBase.java renamed to fluss-client/src/test/java/com/alibaba/fluss/client/admin/ClientToServerITCaseUtils.java

@@ -16,8 +16,6 @@
 
 package com.alibaba.fluss.client.admin;
 
-import com.alibaba.fluss.client.Connection;
-import com.alibaba.fluss.client.ConnectionFactory;
 import com.alibaba.fluss.client.admin.OffsetSpec.LatestSpec;
 import com.alibaba.fluss.client.admin.OffsetSpec.TimestampSpec;
 import com.alibaba.fluss.client.scanner.ScanRecord;
@@ -33,13 +31,8 @@
 import com.alibaba.fluss.metadata.TableDescriptor;
 import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.row.InternalRow;
-import com.alibaba.fluss.server.testutils.FlussClusterExtension;
 import com.alibaba.fluss.types.RowType;
 
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
 import javax.annotation.Nullable;
 
 import java.time.Duration;
@@ -55,48 +48,20 @@
  * The base test class for client to server request and response. The server include
  * CoordinatorServer and TabletServer.
  */
-public abstract class ClientToServerITCaseBase {
-
-    @RegisterExtension
-    public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
-            FlussClusterExtension.builder()
-                    .setNumOfTabletServers(3)
-                    .setClusterConf(initConfig())
-                    .build();
-
-    protected Connection conn;
-    protected Admin admin;
-    protected Configuration clientConf;
-
-    @BeforeEach
-    protected void setup() throws Exception {
-        clientConf = FLUSS_CLUSTER_EXTENSION.getClientConfig();
-        conn = ConnectionFactory.createConnection(clientConf);
-        admin = conn.getAdmin();
-    }
-
-    @AfterEach
-    protected void teardown() throws Exception {
-        if (admin != null) {
-            admin.close();
-            admin = null;
-        }
+public class ClientToServerITCaseUtils {
 
-        if (conn != null) {
-            conn.close();
-            conn = null;
-        }
-    }
-
-    protected long createTable(
-            TablePath tablePath, TableDescriptor tableDescriptor, boolean ignoreIfExists)
+    public static long createTable(
+            Admin admin,
+            TablePath tablePath,
+            TableDescriptor tableDescriptor,
+            boolean ignoreIfExists)
             throws Exception {
         admin.createDatabase(tablePath.getDatabaseName(), ignoreIfExists).get();
         admin.createTable(tablePath, tableDescriptor, ignoreIfExists).get();
         return admin.getTable(tablePath).get().getTableId();
     }
 
-    private static Configuration initConfig() {
+    public static Configuration initConfig() {
         Configuration conf = new Configuration();
         conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 3);
         // set a shorter interval for testing purpose
@@ -109,22 +74,22 @@ private static Configuration initConfig() {
         return conf;
     }
 
-    protected static LogScanner createLogScanner(Table table) {
+    public static LogScanner createLogScanner(Table table) {
         return table.getLogScanner(new LogScan());
     }
 
-    protected static LogScanner createLogScanner(Table table, int[] projectFields) {
+    public static LogScanner createLogScanner(Table table, int[] projectFields) {
         return table.getLogScanner(new LogScan().withProjectedFields(projectFields));
     }
 
-    protected static void subscribeFromBeginning(LogScanner logScanner, Table table) {
+    public static void subscribeFromBeginning(LogScanner logScanner, Table table) {
         int bucketCount = getBucketCount(table);
         for (int i = 0; i < bucketCount; i++) {
             logScanner.subscribeFromBeginning(i);
         }
     }
 
-    protected static void subscribeFromTimestamp(
+    public static void subscribeFromTimestamp(
             PhysicalTablePath physicalTablePath,
             @Nullable Long partitionId,
             Table table,
@@ -147,7 +112,7 @@ protected static void subscribeFromTimestamp(
         }
     }
 
-    protected static void subscribeFromLatestOffset(
+    public static void subscribeFromLatestOffset(
             PhysicalTablePath physicalTablePath,
             @Nullable Long partitionId,
             Table table,
@@ -166,7 +131,7 @@ protected static void subscribeFromLatestOffset(
         }
     }
 
-    protected static List<Integer> getAllBuckets(Table table) {
+    public static List<Integer> getAllBuckets(Table table) {
         List<Integer> buckets = new ArrayList<>();
         int bucketCount = getBucketCount(table);
         for (int i = 0; i < bucketCount; i++) {
@@ -175,7 +140,7 @@ protected static List<Integer> getAllBuckets(Table table) {
         return buckets;
     }
 
-    private static int getBucketCount(Table table) {
+    public static int getBucketCount(Table table) {
         return table.getDescriptor()
                 .getTableDistribution()
                 .flatMap(TableDescriptor.TableDistribution::getBucketCount)
@@ -211,7 +176,7 @@ public static void verifyPartitionLogs(
         verifyRows(rowType, actualRows, expectPartitionsRows);
     }
 
-    protected static void verifyRows(
+    public static void verifyRows(
             RowType rowType,
             Map<Long, List<InternalRow>> actualRows,
             Map<Long, List<InternalRow>> expectedRows) {