This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new e1b09078d [server] Require new leader ack for leader-only rebalance
(#3556)
e1b09078d is described below
commit e1b09078d28c8da1f886e4ba987570991beb82ff
Author: yunhong <[email protected]>
AuthorDate: Thu Jul 9 21:08:05 2026 +0800
[server] Require new leader ack for leader-only rebalance (#3556)
---
.../coordinator/CoordinatorEventProcessor.java | 66 +++++++--
.../coordinator/CoordinatorEventProcessorTest.java | 159 +++++++++++++++++++--
2 files changed, 204 insertions(+), 21 deletions(-)
diff --git
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
index 803b657f4..18c4429af 100644
---
a/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
+++
b/fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java
@@ -1187,7 +1187,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
// server to acknowledge the leader change before proceeding to the
next migration.
for (NotifyLeaderAndIsrResultForBucket
notifyLeaderAndIsrResultForBucket :
notifyLeaderAndIsrResultForBuckets) {
-
tryToCompleteRebalanceTask(notifyLeaderAndIsrResultForBucket.getTableBucket());
+ tryToCompleteRebalanceTask(notifyLeaderAndIsrResultForBucket,
serverId);
}
}
@@ -1600,7 +1600,10 @@ public class CoordinatorEventProcessor implements
EventProcessor {
}
/** try to finish rebalance tasks after receive notify leader and isr
response. */
- private void tryToCompleteRebalanceTask(TableBucket tableBucket) {
+ private void tryToCompleteRebalanceTask(
+ NotifyLeaderAndIsrResultForBucket
notifyLeaderAndIsrResultForBucket,
+ int responseServerId) {
+ TableBucket tableBucket =
notifyLeaderAndIsrResultForBucket.getTableBucket();
RebalancePlanForBucket planForBucket =
rebalanceManager.getRebalancePlanForBucket(tableBucket);
if (planForBucket != null) {
@@ -1609,6 +1612,10 @@ public class CoordinatorEventProcessor implements
EventProcessor {
planForBucket.getOriginReplicas(),
planForBucket.getNewReplicas());
try {
if (planForBucket.isLeaderChanged() &&
!reassignment.isBeingReassigned()) {
+ if (!isSuccessfulLeaderOnlyRebalanceResponseFromNewLeader(
+ notifyLeaderAndIsrResultForBucket,
responseServerId, planForBucket)) {
+ return;
+ }
LeaderAndIsr leaderAndIsr =
zooKeeperClient.getLeaderAndIsr(tableBucket).get();
int currentLeader = leaderAndIsr.leader();
if (currentLeader == planForBucket.getNewLeader()) {
@@ -1616,16 +1623,8 @@ public class CoordinatorEventProcessor implements
EventProcessor {
rebalanceManager.finishRebalanceTask(
tableBucket, RebalanceStatus.COMPLETED);
}
- } else {
- boolean isReassignmentComplete =
- isReassignmentComplete(tableBucket, reassignment);
- if (isReassignmentComplete) {
- LOG.info(
- "Target replicas {} have all caught up with
the leader for reassigning bucket {}",
- reassignment.getTargetReplicas(),
- tableBucket);
- onBucketReassignment(tableBucket, reassignment, true);
- }
+ } else if (notifyLeaderAndIsrResultForBucket.succeeded()) {
+ tryToCompleteReassignmentTask(tableBucket, reassignment);
}
} catch (Exception e) {
LOG.error(
@@ -1635,6 +1634,47 @@ public class CoordinatorEventProcessor implements
EventProcessor {
}
}
+ private void tryToCompleteRebalanceTaskOnLeaderAndIsrChange(TableBucket
tableBucket) {
+ RebalancePlanForBucket planForBucket =
+ rebalanceManager.getRebalancePlanForBucket(tableBucket);
+ if (planForBucket != null) {
+ ReplicaReassignment reassignment =
+ ReplicaReassignment.build(
+ planForBucket.getOriginReplicas(),
planForBucket.getNewReplicas());
+ if (planForBucket.isLeaderChanged() &&
!reassignment.isBeingReassigned()) {
+ return;
+ }
+ try {
+ tryToCompleteReassignmentTask(tableBucket, reassignment);
+ } catch (Exception e) {
+ LOG.error(
+ "Failed to complete the reassignment for table bucket
{}", tableBucket, e);
+ rebalanceManager.finishRebalanceTask(tableBucket,
RebalanceStatus.FAILED);
+ }
+ }
+ }
+
+ private void tryToCompleteReassignmentTask(
+ TableBucket tableBucket, ReplicaReassignment reassignment) throws
Exception {
+ boolean isReassignmentComplete = isReassignmentComplete(tableBucket,
reassignment);
+ if (isReassignmentComplete) {
+ LOG.info(
+ "Target replicas {} have all caught up with the leader for
reassigning bucket {}",
+ reassignment.getTargetReplicas(),
+ tableBucket);
+ onBucketReassignment(tableBucket, reassignment, true);
+ }
+ }
+
+ @VisibleForTesting
+ static boolean isSuccessfulLeaderOnlyRebalanceResponseFromNewLeader(
+ NotifyLeaderAndIsrResultForBucket
notifyLeaderAndIsrResultForBucket,
+ int responseServerId,
+ RebalancePlanForBucket planForBucket) {
+ return notifyLeaderAndIsrResultForBucket.succeeded()
+ && responseServerId == planForBucket.getNewLeader();
+ }
+
/**
* Reassigning replicas for a tableBucket goes through a few steps listed
in the code.
*
@@ -1926,7 +1966,7 @@ public class CoordinatorEventProcessor implements
EventProcessor {
newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
// First, try to judge whether the bucket is in rebalance task when
isr change.
- newLeaderAndIsrList.keySet().forEach(this::tryToCompleteRebalanceTask);
+
newLeaderAndIsrList.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange);
// TODO update metadata for all alive tablet servers.
diff --git
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
index d92953cc2..d3558449f 100644
---
a/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
+++
b/fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java
@@ -1728,13 +1728,13 @@ class CoordinatorEventProcessorTest {
// Set up controlled gateways that capture NotifyLeaderAndIsr calls.
// Gateways start in pass-through mode for table creation, then switch
// to controlled mode to verify sequential leader migration.
- ConcurrentLinkedDeque<CompletableFuture<Void>> pendingTriggers =
+ ConcurrentLinkedDeque<ControlledNotifyTrigger> pendingTriggers =
new ConcurrentLinkedDeque<>();
int[] servers = zookeeperClient.getSortedTabletServerList();
Map<Integer, TabletServerGateway> gateways = new HashMap<>();
ControlledNotifyGateway[] controlledGateways = new
ControlledNotifyGateway[servers.length];
for (int i = 0; i < servers.length; i++) {
- ControlledNotifyGateway gw = new
ControlledNotifyGateway(pendingTriggers);
+ ControlledNotifyGateway gw = new
ControlledNotifyGateway(servers[i], pendingTriggers);
gateways.put(servers[i], gw);
controlledGateways[i] = gw;
}
@@ -1830,6 +1830,101 @@ class CoordinatorEventProcessorTest {
verifyIsr(tb2, 1, Arrays.asList(0, 1, 2));
}
+ @Test
+ void
testLeaderOnlyRebalanceCompletionCheckRequiresSuccessfulResponseFromNewLeader()
{
+ TableBucket tableBucket = new TableBucket(1L, 0);
+ RebalancePlanForBucket planForBucket =
+ new RebalancePlanForBucket(
+ tableBucket, 0, 1, Arrays.asList(0, 1, 2),
Arrays.asList(1, 0, 2));
+ NotifyLeaderAndIsrResultForBucket successResult =
+ new NotifyLeaderAndIsrResultForBucket(tableBucket);
+ NotifyLeaderAndIsrResultForBucket failedResult =
+ new NotifyLeaderAndIsrResultForBucket(
+ tableBucket, new ApiError(Errors.UNKNOWN_SERVER_ERROR,
"failed"));
+
+ assertThat(
+ CoordinatorEventProcessor
+
.isSuccessfulLeaderOnlyRebalanceResponseFromNewLeader(
+ successResult, 1, planForBucket))
+ .isTrue();
+ assertThat(
+ CoordinatorEventProcessor
+
.isSuccessfulLeaderOnlyRebalanceResponseFromNewLeader(
+ successResult, 0, planForBucket))
+ .isFalse();
+ assertThat(
+ CoordinatorEventProcessor
+
.isSuccessfulLeaderOnlyRebalanceResponseFromNewLeader(
+ failedResult, 1, planForBucket))
+ .isFalse();
+ }
+
+ @Test
+ void testLeaderOnlyRebalanceIgnoresSuccessResponseFromOldLeader() throws
Exception {
+ ConcurrentLinkedDeque<ControlledNotifyTrigger> pendingTriggers =
+ new ConcurrentLinkedDeque<>();
+ int[] servers = zookeeperClient.getSortedTabletServerList();
+ Map<Integer, TabletServerGateway> gateways = new HashMap<>();
+ ControlledNotifyGateway[] controlledGateways = new
ControlledNotifyGateway[servers.length];
+ for (int i = 0; i < servers.length; i++) {
+ ControlledNotifyGateway gw = new
ControlledNotifyGateway(servers[i], pendingTriggers);
+ gateways.put(servers[i], gw);
+ controlledGateways[i] = gw;
+ }
+ testCoordinatorChannelManager.setGateways(gateways);
+
+ TablePath t1 = TablePath.of(defaultDatabase,
"test_leader_rebalance_wait_new_leader");
+ Map<Integer, BucketAssignment> bucketAssignments = new HashMap<>();
+ bucketAssignments.put(0, BucketAssignment.of(0, 1, 2));
+ TableAssignment tableAssignment = new
TableAssignment(bucketAssignments);
+ long t1Id =
+ metadataManager.createTable(t1, remoteDataDir, TEST_TABLE,
tableAssignment, false);
+
+ TableBucket tb0 = new TableBucket(t1Id, 0);
+
+ verifyIsr(tb0, 0, Arrays.asList(0, 1, 2));
+
+ for (ControlledNotifyGateway gw : controlledGateways) {
+ gw.enableControlMode();
+ }
+ pendingTriggers.clear();
+
+ Map<TableBucket, RebalancePlanForBucket> rebalancePlan = new
HashMap<>();
+ rebalancePlan.put(
+ tb0,
+ new RebalancePlanForBucket(
+ tb0, 0, 1, Arrays.asList(0, 1, 2), Arrays.asList(1, 0,
2)));
+
+ eventProcessor
+ .getRebalanceManager()
+ .registerRebalance(
+ "rebalance-wait-new-leader-response",
+ rebalancePlan,
+ RebalanceStatus.NOT_STARTED);
+
+ retry(
+ Duration.ofMinutes(1),
+ () -> assertThat(hasPendingNotifyTrigger(pendingTriggers,
0)).isTrue());
+ retry(
+ Duration.ofMinutes(1),
+ () -> assertThat(hasPendingNotifyTrigger(pendingTriggers,
1)).isTrue());
+ assertThat(countInProgressRebalanceTasks(tb0)).isEqualTo(1);
+
+ completePendingNotifyTrigger(pendingTriggers, 0);
+ fromCtx(ctx -> null);
+
+ assertThat(countInProgressRebalanceTasks(tb0)).isEqualTo(1);
+
assertThat(eventProcessor.getRebalanceManager().hasInProgressRebalance()).isTrue();
+
+ completePendingNotifyTrigger(pendingTriggers, 1);
+ retry(
+ Duration.ofMinutes(1),
+ () ->
+
assertThat(eventProcessor.getRebalanceManager().hasInProgressRebalance())
+ .isFalse());
+ verifyIsr(tb0, 1, Arrays.asList(0, 1, 2));
+ }
+
private void verifyIsr(TableBucket tb, int expectedLeader, List<Integer>
expectedIsr)
throws Exception {
LeaderAndIsr leaderAndIsr =
@@ -2231,13 +2326,36 @@ class CoordinatorEventProcessorTest {
}
private static void drainPendingNotifyTriggers(
- ConcurrentLinkedDeque<CompletableFuture<Void>> pendingTriggers) {
- CompletableFuture<Void> trigger;
+ ConcurrentLinkedDeque<ControlledNotifyTrigger> pendingTriggers) {
+ ControlledNotifyTrigger trigger;
while ((trigger = pendingTriggers.poll()) != null) {
trigger.complete(null);
}
}
+ private static boolean hasPendingNotifyTrigger(
+ ConcurrentLinkedDeque<ControlledNotifyTrigger> pendingTriggers,
int responseServerId) {
+ for (ControlledNotifyTrigger trigger : pendingTriggers) {
+ if (trigger.getResponseServerId() == responseServerId) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void completePendingNotifyTrigger(
+ ConcurrentLinkedDeque<ControlledNotifyTrigger> pendingTriggers,
int responseServerId) {
+ for (ControlledNotifyTrigger trigger : pendingTriggers) {
+ if (trigger.getResponseServerId() == responseServerId) {
+ assertThat(pendingTriggers.remove(trigger)).isTrue();
+ trigger.complete(null);
+ return;
+ }
+ }
+ throw new AssertionError(
+ "No pending NotifyLeaderAndIsr response for server " +
responseServerId);
+ }
+
private int countInProgressRebalanceTasks(TableBucket... buckets) {
int count = 0;
for (TableBucket tb : buckets) {
@@ -2275,10 +2393,14 @@ class CoordinatorEventProcessorTest {
*/
private static class ControlledNotifyGateway extends
TestTabletServerGateway {
private volatile boolean controlMode = false;
- private final ConcurrentLinkedDeque<CompletableFuture<Void>>
pendingTriggers;
+ private final int responseServerId;
+ private final ConcurrentLinkedDeque<ControlledNotifyTrigger>
pendingTriggers;
- ControlledNotifyGateway(ConcurrentLinkedDeque<CompletableFuture<Void>>
pendingTriggers) {
+ ControlledNotifyGateway(
+ int responseServerId,
+ ConcurrentLinkedDeque<ControlledNotifyTrigger>
pendingTriggers) {
super(false, Collections.emptySet());
+ this.responseServerId = responseServerId;
this.pendingTriggers = pendingTriggers;
}
@@ -2295,9 +2417,30 @@ class CoordinatorEventProcessorTest {
// Build the proper success response using parent's logic.
NotifyLeaderAndIsrResponse response =
super.notifyLeaderAndIsr(request).join();
// Return a future that completes only when the test releases the
trigger.
- CompletableFuture<Void> trigger = new CompletableFuture<>();
+ ControlledNotifyTrigger trigger = new
ControlledNotifyTrigger(responseServerId);
pendingTriggers.add(trigger);
- return trigger.thenApply(v -> response);
+ return trigger.getFuture().thenApply(v -> response);
+ }
+ }
+
+ private static class ControlledNotifyTrigger {
+ private final int responseServerId;
+ private final CompletableFuture<Void> future = new
CompletableFuture<>();
+
+ ControlledNotifyTrigger(int responseServerId) {
+ this.responseServerId = responseServerId;
+ }
+
+ int getResponseServerId() {
+ return responseServerId;
+ }
+
+ CompletableFuture<Void> getFuture() {
+ return future;
+ }
+
+ void complete(Void value) {
+ future.complete(value);
}
}
