[server] Introduce jitter for upload remote log (apache#872)

Author: luoyuxia

fluss-server/src/main/java/com/alibaba/fluss/server/log/remote/RemoteLogManager.java

@@ -52,6 +52,7 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -283,7 +284,10 @@ private void doHandleLeaderReplica(
                     LOG.info("Created a new remote log task: {} and getting scheduled", task);
                     ScheduledFuture<?> future =
                             rlManagerScheduledThreadPool.scheduleWithFixedDelay(
-                                    task, 0, taskInterval, TimeUnit.MILLISECONDS);
+                                    task,
+                                    Math.abs(ThreadLocalRandom.current().nextLong(taskInterval)),
+                                    taskInterval,
+                                    TimeUnit.MILLISECONDS);
                     return new TaskWithFuture(task, future);
                 });
     }

fluss-server/src/test/java/com/alibaba/fluss/server/log/remote/CommitRemoteLogManifestITCase.java

@@ -28,8 +28,9 @@
 import org.junit.jupiter.api.extension.RegisterExtension;
 
 import java.time.Duration;
-import java.util.Arrays;
+import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import static com.alibaba.fluss.record.TestData.DATA1;
@@ -78,32 +79,44 @@ void testDeleteOutOfSyncReplicaLogAfterCommit() throws Exception {
                     i * 10L);
         }
 
-        // stop a replica to mock follower is out of sync
-        int stopFollower = Stream.of(0, 1, 2).filter(i -> i != leader).findFirst().get();
-        FLUSS_CLUSTER_EXTENSION.stopReplica(stopFollower, tb, 1);
+        // stop replicas to mock followers are out of sync
+        List<Integer> stopFollowers =
+                Stream.of(0, 1, 2).filter(i -> i != leader).collect(Collectors.toList());
+        for (int stopFollower : stopFollowers) {
+            FLUSS_CLUSTER_EXTENSION.stopReplica(stopFollower, tb, 1);
+        }
         leaderGateWay
                 .produceLog(
                         newProduceLogRequest(tableId, 0, -1, genMemoryLogRecordsByObject(DATA1)))
                 .get();
-        FLUSS_CLUSTER_EXTENSION.waitUtilReplicaShrinkFromIsr(tb, stopFollower);
-
-        LogTablet stopfollowerLogTablet =
-                FLUSS_CLUSTER_EXTENSION.waitAndGetFollowerReplica(tb, stopFollower).getLogTablet();
-        assertThat(stopfollowerLogTablet.logSegments()).hasSize(3);
+        for (int stopFollower : stopFollowers) {
+            FLUSS_CLUSTER_EXTENSION.waitUtilReplicaShrinkFromIsr(tb, stopFollower);
+            LogTablet stopfollowerLogTablet =
+                    FLUSS_CLUSTER_EXTENSION
+                            .waitAndGetFollowerReplica(tb, stopFollower)
+                            .getLogTablet();
+            assertThat(stopfollowerLogTablet.logSegments()).hasSize(3);
+        }
 
-        // send notify leader to make remote log tier happen immediately
-        FLUSS_CLUSTER_EXTENSION.notifyLeaderAndIsr(
+        // restart the leader server with a small log tiering interval
+        FLUSS_CLUSTER_EXTENSION.restartTabletServer(
                 leader,
-                DATA1_TABLE_PATH,
-                tb,
-                FLUSS_CLUSTER_EXTENSION.getZooKeeperClient().getLeaderAndIsr(tb).get(),
-                Arrays.asList(0, 1, 2));
+                new Configuration()
+                        .set(
+                                ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION,
+                                Duration.ofMillis(1)));
         FLUSS_CLUSTER_EXTENSION.waitUtilSomeLogSegmentsCopyToRemote(tb);
 
-        // check has two remote log segments for the stopped replica
-        retry(
-                Duration.ofMinutes(1),
-                () -> assertThat(stopfollowerLogTablet.logSegments()).hasSize(2));
+        // check only has two remote log segments for the stopped replicas
+        for (int stopFollower : stopFollowers) {
+            LogTablet stopfollowerLogTablet =
+                    FLUSS_CLUSTER_EXTENSION
+                            .waitAndGetFollowerReplica(tb, stopFollower)
+                            .getLogTablet();
+            retry(
+                    Duration.ofMinutes(1),
+                    () -> assertThat(stopfollowerLogTablet.logSegments()).hasSize(2));
+        }
     }
 
     private static Configuration initConfig() {

fluss-server/src/test/java/com/alibaba/fluss/server/testutils/FlussClusterExtension.java

@@ -273,13 +273,21 @@ public void startTabletServer(int serverId, boolean forceStartIfExists) throws E
                         "Tablet server " + serverId + " already exists.");
             }
         }
+        startTabletServer(serverId, null);
+    }
+
+    private void startTabletServer(int serverId, @Nullable Configuration overwriteConfig)
+            throws Exception {
         String dataDir = getDataDir(serverId);
         Configuration tabletServerConf = new Configuration(clusterConf);
         tabletServerConf.set(ConfigOptions.TABLET_SERVER_ID, serverId);
         tabletServerConf.set(ConfigOptions.DATA_DIR, dataDir);
         tabletServerConf.setString(
                 ConfigOptions.ZOOKEEPER_ADDRESS, zooKeeperServer.getConnectString());
         tabletServerConf.setString(ConfigOptions.BIND_LISTENERS, tabletServerListeners);
+        if (overwriteConfig != null) {
+            tabletServerConf.addAll(overwriteConfig);
+        }
 
         setRemoteDataDir(tabletServerConf);
 
@@ -295,6 +303,11 @@ public void startTabletServer(int serverId, boolean forceStartIfExists) throws E
         tabletServerInfos.put(serverId, serverInfo);
     }
 
+    public void restartTabletServer(int serverId, Configuration overwriteConfig) throws Exception {
+        stopTabletServer(serverId);
+        startTabletServer(serverId, overwriteConfig);
+    }
+
     public void assertHasTabletServerNumber(int tabletServerNumber) {
         CoordinatorGateway coordinatorGateway = newCoordinatorClient();
         retry(