Skip to content

[server] Parallelize replica leader transitions #3999

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zuston wants to merge 10 commits into
base: main
Choose a base branch
from
Open

[server] Parallelize replica leader transitions #3999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
3c52f23
[server] Parallelize replica leader transitions
Aug 14, 2026
7388677
fix(server): await all leader transition futures
Aug 29, 2026
ef4189d
refactor(server): move replica transition executor lifecycle to TabletSe
Aug 29, 2026
9c60225
test(server): cover partial failure in parallel leader election
Aug 29, 2026
3f3423b
test: remove unused ReplicaManager spy assignment
Aug 29, 2026
623b294
fix(server): shut down replica executors in the correct order
Aug 31, 2026
4437fe2
duplicate bucket leaders
Sep 7, 2026
2b0ae5f
make it as 2 concurrency
Sep 7, 2026
74d3ada
remove mockito
Sep 7, 2026
3bb5efc
shutdown fix
Sep 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,7 +405,7 @@ public class ConfigOptions {
public static final ConfigOption<Integer> SERVER_IO_POOL_SIZE =
key("server.io-pool.size")
.intType()
.defaultValue(10)
.defaultValue(2)
.withDescription(
"The size of the IO thread pool to run blocking operations for both coordinator and tablet servers. "
+ "This includes discard unnecessary snapshot files, transfer kv snapshot files, "
Expand Down Expand Up @@ -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)
Comment thread
swuferhong marked this conversation as resolved.

@swuferhong swuferhong Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@zuston zuston Sep 2, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

.withDescription(
"The maximum number of replica role transitions that can run "
+ "concurrently in a TabletServer.");

public static final ConfigOption<Double> TABLET_SERVER_ADVERTISED_RESOURCE_CPU_CORES =
key("tablet-server.advertised-resource.cpu-cores")
.doubleType()
Expand Down
1 change: 1 addition & 0 deletions fluss-common/src/main/java/org/apache/fluss/config/FlussConfigUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,7 @@ public static void validateTabletConfigs(Configuration conf) {
"Configuration %s must be set.", ConfigOptions.TABLET_SERVER_ID.key()));
}
validMinValue(ConfigOptions.TABLET_SERVER_ID, serverId.get(), 0);
validMinValue(conf, ConfigOptions.TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM, 1);
}

public static void validateRemoteDataDirs(Configuration conf) {
Expand Down
8 changes: 8 additions & 0 deletions fluss-common/src/test/java/org/apache/fluss/config/FlussConfigUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,14 @@ void testValidateTabletConfigs() {
.isInstanceOf(IllegalConfigurationException.class)
.hasMessageContaining(ConfigOptions.TABLET_SERVER_ID.key())
.hasMessageContaining("it must be greater than or equal 0");

conf.set(ConfigOptions.TABLET_SERVER_ID, 0);
conf.set(ConfigOptions.TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM, 0);
assertThatThrownBy(() -> validateTabletConfigs(conf))
.isInstanceOf(IllegalConfigurationException.class)
.hasMessageContaining(
ConfigOptions.TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM.key())
.hasMessageContaining("must be greater than or equal 1");
}

@Test
Expand Down
76 changes: 61 additions & 15 deletions fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,12 +145,15 @@
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -196,6 +199,7 @@ public class ReplicaManager implements ServerReconfigurable {

private final TabletServerMetadataCache metadataCache;
private final ExecutorService ioExecutor;
private final ExecutorService replicaTransitionExecutor;
private final ProjectionPushdownCache projectionsCache = new ProjectionPushdownCache();
private final Lock replicaStateChangeLock = new ReentrantLock();

Expand Down Expand Up @@ -259,6 +263,7 @@ public ReplicaManager(
ScannerManager scannerManager,
Clock clock,
ExecutorService ioExecutor,
ExecutorService replicaTransitionExecutor,
LocalDiskManager localDiskManager,
@Nullable PluginManager pluginManager)
throws IOException {
Expand Down Expand Up @@ -287,6 +292,7 @@ public ReplicaManager(
scannerManager,
clock,
ioExecutor,
replicaTransitionExecutor,
localDiskManager,
pluginManager);
}
Expand All @@ -310,6 +316,7 @@ public ReplicaManager(
ScannerManager scannerManager,
Clock clock,
ExecutorService ioExecutor,
ExecutorService replicaTransitionExecutor,
LocalDiskManager localDiskManager,
@Nullable PluginManager pluginManager)
throws IOException {
Expand Down Expand Up @@ -361,6 +368,7 @@ public ReplicaManager(
this.userMetrics = userMetrics;
this.clock = clock;
this.ioExecutor = ioExecutor;
this.replicaTransitionExecutor = replicaTransitionExecutor;
this.minInSyncReplicas = conf.get(ConfigOptions.LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER);
this.scannerManager = checkNotNull(scannerManager, "scannerManager");
// Historical lookup cache capacity currently uses only the first data volume.
Expand Down Expand Up @@ -567,12 +575,18 @@ public void becomeLeaderOrFollower(
inLock(
replicaStateChangeLock,
() -> {
Map<TableBucket, NotifyLeaderAndIsrData> dataByTableBucket =
new LinkedHashMap<>();
for (NotifyLeaderAndIsrData data : notifyLeaderAndIsrDataList) {
dataByTableBucket.put(data.getTableBucket(), data);
}

// check or apply coordinator epoch.
validateAndApplyCoordinatorEpoch(requestCoordinatorEpoch, "notifyLeaderAndIsr");

List<NotifyLeaderAndIsrData> replicasToBeLeader = new ArrayList<>();
List<NotifyLeaderAndIsrData> replicasToBeFollower = new ArrayList<>();
for (NotifyLeaderAndIsrData data : notifyLeaderAndIsrDataList) {
for (NotifyLeaderAndIsrData data : dataByTableBucket.values()) {
LOG.info(
"Try to become leaderAndFollower for {} with isr {}, replicas: {}",
data.getTableBucket(),
Expand Down Expand Up @@ -1389,30 +1403,62 @@ private void makeLeaders(
.map(NotifyLeaderAndIsrData::getTableBucket)
.collect(Collectors.toSet()));

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

@swuferhong swuferhong Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@zuston zuston Sep 2, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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)));
}
}

try {
CompletableFuture.allOf(
makeLeaderFutures.toArray(
new CompletableFuture<?>[makeLeaderFutures.size()]))
.get();
} catch (InterruptedException e) {
makeLeaderFutures.forEach(future -> future.cancel(false));
Thread.currentThread().interrupt();
throw new CompletionException(e);
} catch (ExecutionException e) {
throw new CompletionException(e.getCause());
}
for (CompletableFuture<NotifyLeaderAndIsrResultForBucket> future : makeLeaderFutures) {
NotifyLeaderAndIsrResultForBucket leaderResult = future.join();
Comment thread
swuferhong marked this conversation as resolved.
result.put(leaderResult.getTableBucket(), leaderResult);
}
}

private NotifyLeaderAndIsrResultForBucket makeLeader(
Replica replica, NotifyLeaderAndIsrData data) {
TableBucket tb = data.getTableBucket();
try {
// 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);
return new NotifyLeaderAndIsrResultForBucket(tb);
} catch (Exception e) {
LOG.error("Error make replica {} to leader", tb, e);
return new NotifyLeaderAndIsrResultForBucket(tb, ApiError.fromThrowable(e));
}
}

// NOTE: This method can be removed when fetchFromLake is deprecated
Expand Down
17 changes: 17 additions & 0 deletions fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,9 @@ public class TabletServer extends ServerBase {
@GuardedBy("lock")
private ExecutorService replicaStateChangeExecutor;

@GuardedBy("lock")
private ExecutorService replicaTransitionExecutor;

public TabletServer(Configuration conf) {
this(conf, SystemClock.getInstance());
}
Expand Down Expand Up @@ -285,6 +288,11 @@ protected void startServices() throws Exception {
this.replicaStateChangeExecutor =
Executors.newSingleThreadExecutor(
new ExecutorThreadFactory("tablet-server-replica-state-change"));
this.replicaTransitionExecutor =
Executors.newFixedThreadPool(
conf.get(ConfigOptions.TABLET_SERVER_REPLICA_TRANSITION_THREAD_NUM),
new ExecutorThreadFactory(
"tablet-server-replica-transition-" + serverId));

this.scannerManager = new ScannerManager(conf, scheduler);

Expand All @@ -307,6 +315,7 @@ protected void startServices() throws Exception {
scannerManager,
clock,
ioExecutor,
replicaTransitionExecutor,
localDiskManager,
pluginManager);
replicaManager.startup();
Expand Down Expand Up @@ -478,6 +487,14 @@ CompletableFuture<Void> stopServices() {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}

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

@swuferhong swuferhong Sep 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
} catch (Throwable t) {
exception = ExceptionUtils.firstOrSuppressed(t, exception);
}

try {
if (replicaStateChangeExecutor != null) {
ExecutorUtils.gracefulShutdown(5, TimeUnit.SECONDS, replicaStateChangeExecutor);
Expand Down
Loading
Loading