You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2026/08/14 09:04:09 UTC
[PR] [server] Parallelize replica leader transitions [fluss]
zuston opened a new pull request, #3999:
URL: https://github.com/apache/fluss/pull/3999
### Purpose
This PR speeds up replica leader transitions by processing independent buckets concurrently within a `NotifyLeaderAndIsr` request.
Previously, `ReplicaManager.makeLeaders()` transitioned replicas sequentially. This could significantly increase request latency when multiple replicas became leaders together, especially for primary-key tables where leader initialization may involve restoring RocksDB state from remote snapshots and replaying logs.
### Brief change log
Introduce the option of `tablet-server.replica-transition-thread-num` to control concurrency, with a default value of 10.
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3886184805
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -649,6 +649,14 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`");
+ public static final ConfigOption<Integer> TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM =
+ key("tablet-server.replica-transition-thread-num")
+ .intType()
+ .defaultValue(10)
Review Comment:
The description credits restoring RocksDB from remote snapshots, but initKvTablet already fans that download out over ioExecutor one task per file. Is the win here mainly overlapping one bucket's RocksDB replay with another bucket's download, plus the ZK and manifest reads?
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java:
##########
@@ -1796,6 +1799,63 @@ void becomeLeaderOrFollower() throws Exception {
replicaManager.getReplicaOrException(tb), true, 1, INITIAL_BUCKET_EPOCH);
}
+ @Test
+ void testMakeLeadersInParallel() throws Exception {
+ ReplicaManager spyingReplicaManager = spy(replicaManager);
+ replicaManager = spyingReplicaManager;
Review Comment:
Is this assignment needed? spyingReplicaManager is used everywhere below.
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java:
##########
@@ -1796,6 +1799,63 @@ void becomeLeaderOrFollower() throws Exception {
replicaManager.getReplicaOrException(tb), true, 1, INITIAL_BUCKET_EPOCH);
}
+ @Test
+ void testMakeLeadersInParallel() throws Exception {
+ ReplicaManager spyingReplicaManager = spy(replicaManager);
+ replicaManager = spyingReplicaManager;
+
+ CountDownLatch leadersStarted = new CountDownLatch(2);
+ doAnswer(
+ invocation -> {
+ leadersStarted.countDown();
+ if (!leadersStarted.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError(
+ "Leader transitions did not run in parallel");
+ }
+ return invocation.callRealMethod();
+ })
+ .when(spyingReplicaManager)
+ .makeLeader(any(Replica.class), any(NotifyLeaderAndIsrData.class));
+
+ TableBucket firstBucket = new TableBucket(DATA1_TABLE_ID, 1);
+ TableBucket secondBucket = new TableBucket(DATA1_TABLE_ID, 2);
+ CompletableFuture<List<NotifyLeaderAndIsrResultForBucket>> future =
+ new CompletableFuture<>();
+ spyingReplicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Arrays.asList(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ firstBucket,
+ Arrays.asList(1, 2, 3),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH)),
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ secondBucket,
+ Arrays.asList(1, 2, 3),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH))),
+ future::complete);
+
+ assertThat(future.get())
+ .containsExactlyInAnyOrder(
+ new NotifyLeaderAndIsrResultForBucket(firstBucket),
+ new NotifyLeaderAndIsrResultForBucket(secondBucket));
+ assertThat(spyingReplicaManager.getReplicaOrException(firstBucket).isLeader()).isTrue();
+ assertThat(spyingReplicaManager.getReplicaOrException(secondBucket).isLeader()).isTrue();
+ }
Review Comment:
Could we also cover one bucket failing while the others succeed? That is the part this refactor could break, and pointing a bucket's remote log manifest at a missing file is enough to make registerReplica throw.
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -1389,30 +1398,50 @@ private void makeLeaders(
.map(NotifyLeaderAndIsrData::getTableBucket)
.collect(Collectors.toSet()));
+ List<CompletableFuture<NotifyLeaderAndIsrResultForBucket>> makeLeaderFutures =
+ new ArrayList<>(replicasToBeLeader.size());
for (NotifyLeaderAndIsrData data : replicasToBeLeader) {
TableBucket tb = data.getTableBucket();
try {
Replica replica = getReplicaOrException(tb);
- // register replica to remote log manager first.
- remoteLogManager.registerReplica(replica);
-
- // Load the latest lake progress before leader activation. Historical KV recovery
- // requires its lake log end offset, while failures remain best effort for normal
- // replicas.
- if (replica.isDataLakeEnabled()) {
- updateWithLakeTableSnapshot(replica);
- }
- replica.makeLeader(data);
-
- // start the remote log tiering tasks for leaders
- remoteLogManager.startLogTiering(replica);
- result.put(tb, new NotifyLeaderAndIsrResultForBucket(tb));
+ makeLeaderFutures.add(
+ CompletableFuture.supplyAsync(
+ () -> makeLeader(replica, data), replicaTransitionExecutor));
} catch (Exception e) {
LOG.error("Error make replica {} to leader", tb, e);
result.put(
tb, new NotifyLeaderAndIsrResultForBucket(tb, ApiError.fromThrowable(e)));
}
}
+
+ for (CompletableFuture<NotifyLeaderAndIsrResultForBucket> future : makeLeaderFutures) {
+ NotifyLeaderAndIsrResultForBucket leaderResult = future.join();
Review Comment:
When a task throws something makeLeader does not catch, join() propagates out of makeLeaders and the remaining futures are never joined, so becomeLeaderOrFollower returns and releases replicaStateChangeLock while those transitions are still running. A following stopReplicas can then take the lock against a bucket whose makeLeader has not finished.
Would handling each future at submission time be safer, so the lock is held until every bucket is done?
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -2549,6 +2578,7 @@ public Replica getReplica() {
public static final class OfflineReplica implements HostedReplica {}
public void shutdown() throws InterruptedException {
+ ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaTransitionExecutor);
Review Comment:
shutdownNow() on replicaStateChangeExecutor used to interrupt the thread doing the transition. It now interrupts a thread waiting in join(), so the workers are only interrupted here, after scheduler.shutdown() and scannerManager.close().
Could we create the pool in TabletServer like ioExecutor and drain it at TabletServer:483, so gracefulShutdown reaches the worker threads again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3893372168
##########
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java:
##########
@@ -486,6 +495,14 @@ CompletableFuture<Void> stopServices() {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}
+ try {
+ if (replicaTransitionExecutor != null) {
+ ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaTransitionExecutor);
Review Comment:
drain the transition pool before replicaStateChangeExecutor, since allOf().join() ignores the interrupt and the workers are what need signalling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3886775457
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -649,6 +649,14 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`");
+ public static final ConfigOption<Integer> TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM =
+ key("tablet-server.replica-transition-thread-num")
+ .intType()
+ .defaultValue(10)
Review Comment:
good point! I have updated the PR description.
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java:
##########
@@ -1796,6 +1799,63 @@ void becomeLeaderOrFollower() throws Exception {
replicaManager.getReplicaOrException(tb), true, 1, INITIAL_BUCKET_EPOCH);
}
+ @Test
+ void testMakeLeadersInParallel() throws Exception {
+ ReplicaManager spyingReplicaManager = spy(replicaManager);
+ replicaManager = spyingReplicaManager;
+
+ CountDownLatch leadersStarted = new CountDownLatch(2);
+ doAnswer(
+ invocation -> {
+ leadersStarted.countDown();
+ if (!leadersStarted.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError(
+ "Leader transitions did not run in parallel");
+ }
+ return invocation.callRealMethod();
+ })
+ .when(spyingReplicaManager)
+ .makeLeader(any(Replica.class), any(NotifyLeaderAndIsrData.class));
+
+ TableBucket firstBucket = new TableBucket(DATA1_TABLE_ID, 1);
+ TableBucket secondBucket = new TableBucket(DATA1_TABLE_ID, 2);
+ CompletableFuture<List<NotifyLeaderAndIsrResultForBucket>> future =
+ new CompletableFuture<>();
+ spyingReplicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Arrays.asList(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ firstBucket,
+ Arrays.asList(1, 2, 3),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH)),
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ secondBucket,
+ Arrays.asList(1, 2, 3),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH))),
+ future::complete);
+
+ assertThat(future.get())
+ .containsExactlyInAnyOrder(
+ new NotifyLeaderAndIsrResultForBucket(firstBucket),
+ new NotifyLeaderAndIsrResultForBucket(secondBucket));
+ assertThat(spyingReplicaManager.getReplicaOrException(firstBucket).isLeader()).isTrue();
+ assertThat(spyingReplicaManager.getReplicaOrException(secondBucket).isLeader()).isTrue();
+ }
Review Comment:
added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3910454375
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -649,6 +649,14 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`");
+ public static final ConfigOption<Integer> TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM =
+ key("tablet-server.replica-transition-thread-num")
+ .intType()
+ .defaultValue(10)
Review Comment:
Makes sense. For now, we could use a more conservative default. How about setting it to 2?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3886186738
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java:
##########
@@ -1796,6 +1799,63 @@ void becomeLeaderOrFollower() throws Exception {
replicaManager.getReplicaOrException(tb), true, 1, INITIAL_BUCKET_EPOCH);
}
+ @Test
+ void testMakeLeadersInParallel() throws Exception {
+ ReplicaManager spyingReplicaManager = spy(replicaManager);
+ replicaManager = spyingReplicaManager;
+
+ CountDownLatch leadersStarted = new CountDownLatch(2);
+ doAnswer(
+ invocation -> {
+ leadersStarted.countDown();
+ if (!leadersStarted.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError(
+ "Leader transitions did not run in parallel");
+ }
+ return invocation.callRealMethod();
+ })
+ .when(spyingReplicaManager)
+ .makeLeader(any(Replica.class), any(NotifyLeaderAndIsrData.class));
+
+ TableBucket firstBucket = new TableBucket(DATA1_TABLE_ID, 1);
+ TableBucket secondBucket = new TableBucket(DATA1_TABLE_ID, 2);
+ CompletableFuture<List<NotifyLeaderAndIsrResultForBucket>> future =
+ new CompletableFuture<>();
+ spyingReplicaManager.becomeLeaderOrFollower(
+ INITIAL_COORDINATOR_EPOCH,
+ Arrays.asList(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ firstBucket,
+ Arrays.asList(1, 2, 3),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH)),
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ secondBucket,
+ Arrays.asList(1, 2, 3),
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ Arrays.asList(1, 2, 3),
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH))),
+ future::complete);
+
+ assertThat(future.get())
+ .containsExactlyInAnyOrder(
+ new NotifyLeaderAndIsrResultForBucket(firstBucket),
+ new NotifyLeaderAndIsrResultForBucket(secondBucket));
+ assertThat(spyingReplicaManager.getReplicaOrException(firstBucket).isLeader()).isTrue();
+ assertThat(spyingReplicaManager.getReplicaOrException(secondBucket).isLeader()).isTrue();
+ }
Review Comment:
Could we also cover one bucket failing while the others succeed? That is the part this refactor could break, and a stale bucket epoch is enough to make Replica.makeLeader throw for just that bucket.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "github-actions[bot] (via GitHub) via issues" <is...@fluss.apache.org>.
github-actions[bot] commented on PR #3999:
URL: https://github.com/apache/fluss/pull/3999#issuecomment-6680758635
This PR has been marked as stale because it has had no activity for 60 days. It will be closed in 10 days unless it receives a new comment, commit or other update.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3910454375
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -649,6 +649,14 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`");
+ public static final ConfigOption<Integer> TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM =
+ key("tablet-server.replica-transition-thread-num")
+ .intType()
+ .defaultValue(10)
Review Comment:
Makes sense. For now, we could use a more conservative default. How about setting it to 1 to keep it consistent with the previous behavior?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3886269822
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -1389,30 +1398,50 @@ private void makeLeaders(
.map(NotifyLeaderAndIsrData::getTableBucket)
.collect(Collectors.toSet()));
+ List<CompletableFuture<NotifyLeaderAndIsrResultForBucket>> makeLeaderFutures =
+ new ArrayList<>(replicasToBeLeader.size());
for (NotifyLeaderAndIsrData data : replicasToBeLeader) {
TableBucket tb = data.getTableBucket();
try {
Replica replica = getReplicaOrException(tb);
- // register replica to remote log manager first.
- remoteLogManager.registerReplica(replica);
-
- // Load the latest lake progress before leader activation. Historical KV recovery
- // requires its lake log end offset, while failures remain best effort for normal
- // replicas.
- if (replica.isDataLakeEnabled()) {
- updateWithLakeTableSnapshot(replica);
- }
- replica.makeLeader(data);
-
- // start the remote log tiering tasks for leaders
- remoteLogManager.startLogTiering(replica);
- result.put(tb, new NotifyLeaderAndIsrResultForBucket(tb));
+ makeLeaderFutures.add(
+ CompletableFuture.supplyAsync(
+ () -> makeLeader(replica, data), replicaTransitionExecutor));
} catch (Exception e) {
LOG.error("Error make replica {} to leader", tb, e);
result.put(
tb, new NotifyLeaderAndIsrResultForBucket(tb, ApiError.fromThrowable(e)));
}
}
+
+ for (CompletableFuture<NotifyLeaderAndIsrResultForBucket> future : makeLeaderFutures) {
+ NotifyLeaderAndIsrResultForBucket leaderResult = future.join();
Review Comment:
TabletManagerBase.closeTabletsConcurrently uses allOf so every future completes before it returns. Joining in a loop means an escaping Error leaves the remaining transitions running with replicaStateChangeLock already released.
Could we follow the same pattern?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3910052032
##########
fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java:
##########
@@ -649,6 +649,14 @@ public class ConfigOptions {
"The rack for the tabletServer. This will be used in rack aware bucket assignment "
+ "for fault tolerance. Examples: `RACK1`, `cn-hangzhou-server10`");
+ public static final ConfigOption<Integer> TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM =
+ key("tablet-server.replica-transition-thread-num")
+ .intType()
+ .defaultValue(10)
Review Comment:
The configured value only bounds the number of bucket transitions; it does not bound the actual recovery concurrency.
Each KV leader recovery creates a private `RemoteLogFetcher` executor, whose default size is `3`. With the proposed default of `10` transitions, a mass leader election may therefore create up to `30` remote-log download threads, in addition to the `10` transition threads, shared snapshot-transfer work, and RocksDB background initialization/compaction threads.
This burst happens precisely during failover or rebalancing, when the server is already under pressure, and may amplify CPU, native-memory, file-descriptor, disk-I/O, and remote-storage load. Ten concurrent RocksDB initializations are also potentially expensive even when remote-log replay is not required.
Could we either:
- use a substantially more conservative default, such as 1 or 2;
- share and globally bound the remote-log recovery executor instead of creating one per bucket; or
provide a multi-bucket recovery benchmark/stress test demonstrating that the default of 10 is safe?
I think this should be evaluated before making 10 the production default. WDYT @platinumhamburg
##########
fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java:
##########
@@ -478,6 +487,14 @@ CompletableFuture<Void> stopServices() {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}
+ try {
+ if (replicaTransitionExecutor != null) {
+ ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaTransitionExecutor);
Review Comment:
`ExecutorUtils.gracefulShutdown()` eventually calls `shutdownNow()`, but queued `CompletableFuture$AsyncSupply` tasks returned by `shutdownNow()` are only discarded; their corresponding futures are not automatically completed or cancelled.
If the pool is saturated, a running transition exceeds the five-second timeout, and other transitions remain queued, `makeLeaders()` can stay blocked forever in `CompletableFuture.allOf(...).join()`. Shutting down `replicaStateChangeExecutor` afterwards cannot release that thread because `join()` is not interruptible.
Meanwhile,` ReplicaManager` and tablet resources continue shutting down and may race with transition workers that are still running.
Could we explicitly track and cancel the transition futures during forced shutdown, use an interruptible wait, and ensure every submitted future reaches a terminal state? Please also add a deterministic shutdown test with a one-thread transition pool, one blocked transition, and a second queued transition.
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -1389,30 +1394,54 @@ private void makeLeaders(
.map(NotifyLeaderAndIsrData::getTableBucket)
.collect(Collectors.toSet()));
+ List<CompletableFuture<NotifyLeaderAndIsrResultForBucket>> makeLeaderFutures =
+ new ArrayList<>(replicasToBeLeader.size());
for (NotifyLeaderAndIsrData data : replicasToBeLeader) {
Review Comment:
Can we guarantee or validate that `replicasToBeLeader `contains each `TableBucket` only once?
The normal coordinator path currently builds the request from a map, but the RPC representation is a repeated list and the `TabletServer` does not enforce uniqueness. Duplicate entries would schedule concurrent transitions for the same Replica.
Replica.makeLeader() is protected by its per-replica lock, but `registerReplica()` and the lake-snapshot refresh run before that lock, so their side effects can still race.
Please either reject duplicate buckets, deduplicate the request before submission, or serialize transition tasks by `TableBucket`.
##########
fluss-server/src/test/java/org/apache/fluss/server/replica/ReplicaManagerTest.java:
##########
@@ -1796,6 +1799,96 @@ void becomeLeaderOrFollower() throws Exception {
replicaManager.getReplicaOrException(tb), true, 1, INITIAL_BUCKET_EPOCH);
}
+ @Test
+ void testMakeLeadersInParallelWithPartialFailure() throws Exception {
+ TableBucket firstBucket = new TableBucket(DATA1_TABLE_ID, 1);
+ TableBucket secondBucket = new TableBucket(DATA1_TABLE_ID, 2);
+ List<Integer> replicas = Arrays.asList(1, 2, 3);
+ makeLogTableAsLeader(firstBucket.getBucket());
+ replicaManager
+ .getReplicaOrException(firstBucket)
+ .makeLeader(
+ new NotifyLeaderAndIsrData(
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ firstBucket,
+ replicas,
+ new LeaderAndIsr(
+ TABLET_SERVER_ID,
+ INITIAL_LEADER_EPOCH,
+ replicas,
+ Collections.emptyList(),
+ INITIAL_COORDINATOR_EPOCH,
+ INITIAL_BUCKET_EPOCH + 1)));
+
+ ReplicaManager spyingReplicaManager = spy(replicaManager);
+
+ CountDownLatch leadersStarted = new CountDownLatch(2);
+ doAnswer(
+ invocation -> {
+ leadersStarted.countDown();
+ if (!leadersStarted.await(10, TimeUnit.SECONDS)) {
+ throw new AssertionError(
+ "Leader transitions did not run in parallel");
+ }
+ return invocation.callRealMethod();
+ })
+ .when(spyingReplicaManager)
Review Comment:
Could we avoid `Mockito` for test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #3999:
URL: https://github.com/apache/fluss/pull/3999#discussion_r3910508880
##########
fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java:
##########
@@ -1389,30 +1394,54 @@ private void makeLeaders(
.map(NotifyLeaderAndIsrData::getTableBucket)
.collect(Collectors.toSet()));
+ List<CompletableFuture<NotifyLeaderAndIsrResultForBucket>> makeLeaderFutures =
+ new ArrayList<>(replicasToBeLeader.size());
for (NotifyLeaderAndIsrData data : replicasToBeLeader) {
Review Comment:
Good point, this is necessary to ensure the unique before entering the parallelize transitions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Parallelize replica leader transitions [fluss]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #3999:
URL: https://github.com/apache/fluss/pull/3999#issuecomment-5292394432
could you help review this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org