[server] Parallelize replica leader transitions by zuston · Pull Request #3999 · apache/fluss

zuston · 2026-08-14T09:04:06Z

Purpose

This PR reduces the latency of replica leader transitions by processing independent buckets concurrently within a NotifyLeaderAndIsr request.
Previously, ReplicaManager.makeLeaders() transitioned replicas sequentially. This increased request latency when multiple replicas became leaders together, especially for primary-key tables where each transition may initialize RocksDB by restoring a remote snapshot and replaying logs.

Snapshot files within a single bucket are already downloaded concurrently through ioExecutor. The main benefit of this change is parallelizing the bucket-level transition: independent buckets can overlap snapshot download, RocksDB initialization and replay, snapshot metadata reads, and remote-log setup.

Brief change log

Introduce the tablet-server.replica-transition-thread-num option to control concurrency, with a default value of 10.

In this first phase, makeFollowers remains on the existing state-change executor and is not moved into the background transition pool.

Tests

API and Format

Documentation

zuston · 2026-08-14T10:42:58Z

could you help review this? @fresh-borzoni

fresh-borzoni

@zuston Thank you for the PR, left some comments and questions, PTAL

fresh-borzoni · 2026-08-29T09:44:26Z

+                        new NotifyLeaderAndIsrResultForBucket(secondBucket));
+        assertThat(spyingReplicaManager.getReplicaOrException(firstBucket).isLeader()).isTrue();
+        assertThat(spyingReplicaManager.getReplicaOrException(secondBucket).isLeader()).isTrue();
+    }


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.

fresh-borzoni · 2026-08-29T09:45:22Z

+    @Test
+    void testMakeLeadersInParallel() throws Exception {
+        ReplicaManager spyingReplicaManager = spy(replicaManager);
+        replicaManager = spyingReplicaManager;


Is this assignment needed? spyingReplicaManager is used everywhere below.

fresh-borzoni · 2026-08-29T10:18:34Z

    public static final class OfflineReplica implements HostedReplica {}

    public void shutdown() throws InterruptedException {
+        ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaTransitionExecutor);


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?

fresh-borzoni

@zuston Thank you, LGTM overall, one comment, PTAL

fresh-borzoni · 2026-08-31T09:49:17Z


+            try {
+                if (replicaTransitionExecutor != null) {
+                    ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaTransitionExecutor);


drain the transition pool before replicaStateChangeExecutor, since allOf().join() ignores the interrupt and the workers are what need signalling.

fresh-borzoni

@zuston Thank you, LGTM 👍

cc @swuferhong

swuferhong

Hi, @zuston, thanks for your contributions, I left some comments:

swuferhong · 2026-09-02T01:58:21Z

+    public static final ConfigOption<Integer> TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM =
+            key("tablet-server.replica-transition-thread-num")
+                    .intType()
+                    .defaultValue(10)


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

Makes sense. For now, we could use a more conservative default. How about setting it to 2?

swuferhong · 2026-09-02T02:00:48Z


+            try {
+                if (replicaTransitionExecutor != null) {
+                    ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaTransitionExecutor);


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.

swuferhong · 2026-09-02T02:01:46Z

+                            }
+                            return invocation.callRealMethod();
+                        })
+                .when(spyingReplicaManager)


Could we avoid Mockito for test?

swuferhong · 2026-09-02T02:04:09Z


+        List<CompletableFuture<NotifyLeaderAndIsrResultForBucket>> makeLeaderFutures =
+                new ArrayList<>(replicasToBeLeader.size());
        for (NotifyLeaderAndIsrData data : replicasToBeLeader) {


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.

Good point, this is necessary to ensure the unique before entering the parallelize transitions

[server] Parallelize replica leader transitions

3c52f23

zuston force-pushed the makeleader branch from c27272e to 3c52f23 Compare August 28, 2026 08:47

fresh-borzoni reviewed Aug 29, 2026

View reviewed changes

zhangjunfan added 4 commits August 29, 2026 21:49

fix(server): await all leader transition futures

7388677

refactor(server): move replica transition executor lifecycle to TabletSe

ef4189d

test(server): cover partial failure in parallel leader election

9c60225

test: remove unused ReplicaManager spy assignment

3f3423b

zuston requested a review from fresh-borzoni August 29, 2026 14:09

fresh-borzoni reviewed Aug 31, 2026

View reviewed changes

fix(server): shut down replica executors in the correct order

623b294

zuston requested a review from fresh-borzoni August 31, 2026 14:40

fresh-borzoni reviewed Aug 31, 2026

View reviewed changes

fresh-borzoni approved these changes Aug 31, 2026

View reviewed changes

swuferhong self-requested a review September 2, 2026 01:14

swuferhong reviewed Sep 2, 2026

View reviewed changes

zhangjunfan added 4 commits September 7, 2026 15:26

duplicate bucket leaders

4437fe2

make it as 2 concurrency

2b0ae5f

remove mockito

74d3ada

shutdown fix

3bb5efc

Uh oh!

Conversation

zuston commented Aug 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Brief change log

Tests

API and Format

Documentation

Uh oh!

zuston commented Aug 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

fresh-borzoni Aug 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fresh-borzoni left a comment

Choose a reason for hiding this comment

Uh oh!

swuferhong left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zuston Sep 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

zuston commented Aug 14, 2026 •

edited

Loading

zuston commented Aug 14, 2026 •

edited

Loading

fresh-borzoni Aug 29, 2026 •

edited

Loading

zuston Sep 2, 2026 •

edited

Loading