You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fhan688 (via GitHub)" <gi...@apache.org> on 2026/07/08 12:42:27 UTC
[PR] [server] Add in-memory rebalance round limit [fluss]
fhan688 opened a new pull request, #3622:
URL: https://github.com/apache/fluss/pull/3622
### Purpose
Linked issue: https://github.com/apache/fluss/issues/3348 https://github.com/apache/fluss/issues/3536
This PR is derived from the previous rebalance throttling work in #3498 and #3537.
#3537 adds `coordinator.rebalance.max-inflight-tasks` to control how many bucket-level rebalance tasks can run concurrently. #3498 explored limiting large rebalance plans by splitting them into rounds, but that approach introduced new ZooKeeper metadata models, serde, and znodes.
This PR keeps the round-limiting capability from #3498, but reworks it on top of #3537 with a lower-intrusion design: the coordinator uses an in-memory round/window inside `RebalanceManager` and does not add any new ZooKeeper model, serde, or znode.
The default behavior remains unchanged.
### Brief change log
- Add `coordinator.rebalance.max-buckets-per-round` with default value `0`.
- `0` disables round limiting.
- A positive value limits how many bucket-level rebalance tasks are admitted into one in-memory scheduling round.
- Extend `RebalanceManager` with an in-memory rebalance round/window.
- Keep the full rebalance plan persisted in the existing legacy `RebalanceTask` znode.
- Sort bucket tasks deterministically by `tableId`, `partitionId`, and `bucketId`.
- Admit at most `coordinator.rebalance.max-buckets-per-round` bucket tasks into the current round.
- Open the next round only after the current round has no pending or in-flight bucket tasks.
- Continue using `coordinator.rebalance.max-inflight-tasks` from #3537 to control concurrent execution within the active round.
- Keep the implementation ZooKeeper-compatible with the existing rebalance flow.
- No new ZooKeeper metadata model.
- No new JSON serde.
- No new znode under the rebalance path.
- Recovery still uses the existing persisted `RebalanceTask` and rebuilds the in-memory scheduling window after CoordinatorServer restart.
- Update rebalance documentation to explain how:
- `coordinator.rebalance.max-inflight-tasks` controls concurrent bucket movements.
- `coordinator.rebalance.max-buckets-per-round` controls per-round admission.
- Very small round sizes, such as `1`, can create many small scheduling rounds and should be used only when strict serial admission is required.
### Tests
- `git diff --check`
- `mvn -pl fluss-server -am -DskipITs -Dcheckstyle.skip=true -DfailIfNoTests=false -Dtest=RebalanceManagerTest test`
### API and Format
- Adds a new coordinator configuration:
- `coordinator.rebalance.max-buckets-per-round`
- No public client API changes.
- No RPC protocol changes.
- No storage format changes.
- No new ZooKeeper metadata, serde, or znode.
Backward compatibility:
- The default value `0` preserves the existing behavior.
- Existing rebalance task metadata remains the only persisted rebalance execution format.
- `coordinator.rebalance.max-inflight-tasks` keeps the #3537 behavior and remains responsible for concurrent execution throttling.
### Documentation
- Add `coordinator.rebalance.max-buckets-per-round` to `website/docs/maintenance/configuration.md`.
- Update `website/docs/maintenance/operations/rebalance.md` with operational guidance for using `coordinator.rebalance.max-inflight-tasks` and `coordinator.rebalance.max-buckets-
per-round` together.
--
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] Add in-memory rebalance round limit [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3622:
URL: https://github.com/apache/fluss/pull/3622#discussion_r3548338709
##########
fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java:
##########
@@ -110,6 +112,8 @@ class DynamicServerConfig {
void register(ServerReconfigurable serverReconfigurable) {
serverReconfigures.put(serverReconfigurable.getClass(), serverReconfigurable);
+ serverReconfigurable.validate(currentConfig);
+ serverReconfigurable.reconfigure(currentConfig);
}
Review Comment:
`register()` mutates and reads `currentConfig` without the `ReadWriteLock`, even though `updateDynamicConfig()` updates `currentConfig` under the write lock. This creates a data race (late registration can observe a stale/inconsistent `currentConfig`) and can also leave a failing `ServerReconfigurable` in `serverReconfigures` because it is added to the map before `validate()` succeeds.
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java:
##########
@@ -2241,7 +2241,14 @@ private static void drainPendingNotifyTriggers(
private int countInProgressRebalanceTasks(TableBucket... buckets) {
int count = 0;
for (TableBucket tb : buckets) {
- if (eventProcessor.getRebalanceManager().getRebalancePlanForBucket(tb) != null) {
+ RebalanceStatus status =
+ eventProcessor
+ .getRebalanceManager()
+ .listRebalanceProgress(null)
+ .progressForBucketMap()
+ .get(tb)
+ .status();
+ if (!RebalanceStatus.FINAL_STATUSES.contains(status)) {
count++;
}
Review Comment:
`countInProgressRebalanceTasks(...)` assumes `listRebalanceProgress(null)` is non-null and that `progressForBucketMap().get(tb)` returns a non-null entry. If a bucket is absent (e.g., after cancel or if the progress map is cleared), this will throw an NPE and make the test brittle.
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java:
##########
@@ -308,6 +349,385 @@ void testTimeoutTreatsTaskAsCompleted() throws Exception {
manager.close();
}
+ @Test
+ void testRegisterRebalanceRespectsMaxInflightTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("inflight-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("inflight-test", plan, NOT_STARTED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.close();
+ }
+
+ @Test
+ void testRebalanceRoundLimitsActivatedBuckets() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(new RebalanceTask("round-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("round-test", plan, NOT_STARTED);
+
+ assertThat(manager.getMaxBucketsPerRound()).isEqualTo(2);
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(1), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(4);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(eventProcessor.executedPlans.get(3).getTableBucket()).isEqualTo(buckets.get(3));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testRebalanceRoundWorksWithLowerMaxInflightTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 1);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(3);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("round-lower-inflight-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("round-lower-inflight-test", plan, NOT_STARTED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+ assertThat(eventProcessor.executedPlans.get(0).getTableBucket()).isEqualTo(buckets.get(0));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(eventProcessor.executedPlans.get(1).getTableBucket()).isEqualTo(buckets.get(1));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.finishRebalanceTask(buckets.get(1), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isZero();
+
+ manager.close();
+ }
+
+ @Test
+ void testZeroMaxBucketsPerRoundKeepsMaxInflightBehavior() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 0);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("unlimited-round-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("unlimited-round-test", plan, NOT_STARTED);
+
+ assertThat(manager.getMaxBucketsPerRound()).isZero();
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.close();
+ }
+
+ @Test
+ void testMaxBucketsPerRoundSmallerThanMaxInflightCapsScheduling() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 3);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 1);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(3);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("round-smaller-than-inflight-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("round-smaller-than-inflight-test", plan, NOT_STARTED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+ assertThat(eventProcessor.executedPlans.get(0).getTableBucket()).isEqualTo(buckets.get(0));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(eventProcessor.executedPlans.get(1).getTableBucket()).isEqualTo(buckets.get(1));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testNegativeMaxBucketsPerRoundIsRejected() {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, -1);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(new Configuration());
+
+ assertThatThrownBy(
+ () ->
+ new RebalanceManager(
+ eventProcessor,
+ zookeeperClient,
+ eventManager,
+ clock,
+ conf,
+ executor))
+ .hasMessageContaining(
+ ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND.key());
+ }
+
+ @Test
+ void testIncreaseMaxInflightRebalanceTasksStartsMoreTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 1);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(4);
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("increase-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("increase-inflight-test", plan, NOT_STARTED);
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 3);
+ manager.reconfigure(newConfig);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(1);
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 3);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(3);
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(3);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testZeroMaxInflightRebalanceTasksPausesAndResumesScheduling() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 0);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(4);
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("paused-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("paused-inflight-test", plan, NOT_STARTED);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isZero();
+ assertThat(eventProcessor.executedPlans).isEmpty();
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(4);
+ assertThat(manager.hasInProgressRebalance()).isTrue();
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ manager.reconfigure(newConfig);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isZero();
+ assertThat(eventProcessor.executedPlans).isEmpty();
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 2);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(2);
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.close();
+ }
+
+ @Test
+ void testDecreaseMaxInflightRebalanceTasksToZeroDoesNotCancelRunningTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 3);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("decrease-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("decrease-inflight-test", plan, NOT_STARTED);
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 0);
+ manager.reconfigure(newConfig);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(3);
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 0);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isZero();
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+ manager.finishRebalanceTask(buckets.get(1), COMPLETED);
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+
+ manager.finishRebalanceTask(buckets.get(2), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isZero();
+
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 1);
+ manager.reconfigure(newConfig);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 1);
+
+ assertThat(eventProcessor.executedPlans).hasSize(4);
+ assertThat(eventProcessor.executedPlans.get(3).getTableBucket()).isEqualTo(buckets.get(3));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testTimeoutEnqueuesEventsForAllInflightTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(3);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("timeout-all-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("timeout-all-inflight-test", plan, NOT_STARTED);
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+
+ clock.advanceTime(Duration.ofMillis(130_000));
+ manager.checkTimeout();
+
+ assertThat(eventManager.events).hasSize(2);
+ assertThat(((RebalanceTaskTimeoutEvent) eventManager.events.get(0)).getTableBucket())
+ .isEqualTo(buckets.get(0));
+ assertThat(((RebalanceTaskTimeoutEvent) eventManager.events.get(1)).getTableBucket())
+ .isEqualTo(buckets.get(1));
Review Comment:
This test assumes `checkTimeout()` enqueues timeout events in a stable order (bucket 0 then bucket 1). After the change, `checkTimeout()` iterates a `HashMap` snapshot of `inflightRebalanceTaskStartMs`, so the enqueue order is not deterministic when multiple tasks are in-flight; this can make the test flaky.
--
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] Add in-memory rebalance round limit [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on code in PR #3622:
URL: https://github.com/apache/fluss/pull/3622#discussion_r3548702323
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java:
##########
@@ -2241,7 +2241,14 @@ private static void drainPendingNotifyTriggers(
private int countInProgressRebalanceTasks(TableBucket... buckets) {
int count = 0;
for (TableBucket tb : buckets) {
- if (eventProcessor.getRebalanceManager().getRebalancePlanForBucket(tb) != null) {
+ RebalanceStatus status =
+ eventProcessor
+ .getRebalanceManager()
+ .listRebalanceProgress(null)
+ .progressForBucketMap()
+ .get(tb)
+ .status();
+ if (!RebalanceStatus.FINAL_STATUSES.contains(status)) {
count++;
}
Review Comment:
Good point. Fixed the test helper to handle both a null RebalanceProgress and missing bucket progress entries. Missing progress is now treated as not in progress instead of throwing an NPE.
--
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] Add in-memory rebalance round limit [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on PR #3622:
URL: https://github.com/apache/fluss/pull/3622#issuecomment-4916280102
This PR is derived from the previous rebalance throttling work in https://github.com/apache/fluss/pull/3498 and https://github.com/apache/fluss/pull/3537, please help review, thanks! @swuferhong @wuchong
--
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] Add in-memory rebalance round limit [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 closed pull request #3622: [server] Add in-memory rebalance round limit
URL: https://github.com/apache/fluss/pull/3622
--
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] Add in-memory rebalance round limit [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on code in PR #3622:
URL: https://github.com/apache/fluss/pull/3622#discussion_r3548703262
##########
fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java:
##########
@@ -308,6 +349,385 @@ void testTimeoutTreatsTaskAsCompleted() throws Exception {
manager.close();
}
+ @Test
+ void testRegisterRebalanceRespectsMaxInflightTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("inflight-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("inflight-test", plan, NOT_STARTED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.close();
+ }
+
+ @Test
+ void testRebalanceRoundLimitsActivatedBuckets() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(new RebalanceTask("round-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("round-test", plan, NOT_STARTED);
+
+ assertThat(manager.getMaxBucketsPerRound()).isEqualTo(2);
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(1), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(4);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(eventProcessor.executedPlans.get(3).getTableBucket()).isEqualTo(buckets.get(3));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testRebalanceRoundWorksWithLowerMaxInflightTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 1);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(3);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("round-lower-inflight-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("round-lower-inflight-test", plan, NOT_STARTED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+ assertThat(eventProcessor.executedPlans.get(0).getTableBucket()).isEqualTo(buckets.get(0));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(eventProcessor.executedPlans.get(1).getTableBucket()).isEqualTo(buckets.get(1));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.finishRebalanceTask(buckets.get(1), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isZero();
+
+ manager.close();
+ }
+
+ @Test
+ void testZeroMaxBucketsPerRoundKeepsMaxInflightBehavior() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 0);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("unlimited-round-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("unlimited-round-test", plan, NOT_STARTED);
+
+ assertThat(manager.getMaxBucketsPerRound()).isZero();
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(3);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(eventProcessor.executedPlans.get(2).getTableBucket()).isEqualTo(buckets.get(2));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.close();
+ }
+
+ @Test
+ void testMaxBucketsPerRoundSmallerThanMaxInflightCapsScheduling() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 3);
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, 1);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(3);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("round-smaller-than-inflight-test", NOT_STARTED, plan));
+
+ manager.registerRebalance("round-smaller-than-inflight-test", plan, NOT_STARTED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+ assertThat(eventProcessor.executedPlans.get(0).getTableBucket()).isEqualTo(buckets.get(0));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(eventProcessor.executedPlans.get(1).getTableBucket()).isEqualTo(buckets.get(1));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testNegativeMaxBucketsPerRoundIsRejected() {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND, -1);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(new Configuration());
+
+ assertThatThrownBy(
+ () ->
+ new RebalanceManager(
+ eventProcessor,
+ zookeeperClient,
+ eventManager,
+ clock,
+ conf,
+ executor))
+ .hasMessageContaining(
+ ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND.key());
+ }
+
+ @Test
+ void testIncreaseMaxInflightRebalanceTasksStartsMoreTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 1);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(4);
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("increase-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("increase-inflight-test", plan, NOT_STARTED);
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 3);
+ manager.reconfigure(newConfig);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(1);
+ assertThat(eventProcessor.executedPlans).hasSize(1);
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 3);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(3);
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(3);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testZeroMaxInflightRebalanceTasksPausesAndResumesScheduling() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 0);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(4);
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("paused-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("paused-inflight-test", plan, NOT_STARTED);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isZero();
+ assertThat(eventProcessor.executedPlans).isEmpty();
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(4);
+ assertThat(manager.hasInProgressRebalance()).isTrue();
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ manager.reconfigure(newConfig);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isZero();
+ assertThat(eventProcessor.executedPlans).isEmpty();
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 2);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(2);
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+
+ manager.close();
+ }
+
+ @Test
+ void testDecreaseMaxInflightRebalanceTasksToZeroDoesNotCancelRunningTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 3);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(5);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("decrease-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("decrease-inflight-test", plan, NOT_STARTED);
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+
+ Configuration newConfig = new Configuration(conf);
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 0);
+ manager.reconfigure(newConfig);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isEqualTo(3);
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 0);
+
+ assertThat(manager.getMaxInflightRebalanceTasks()).isZero();
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ manager.finishRebalanceTask(buckets.get(0), COMPLETED);
+ manager.finishRebalanceTask(buckets.get(1), COMPLETED);
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+
+ manager.finishRebalanceTask(buckets.get(2), COMPLETED);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ assertThat(countStatus(manager, RebalanceStatus.NOT_STARTED)).isEqualTo(2);
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isZero();
+
+ newConfig.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 1);
+ manager.reconfigure(newConfig);
+
+ assertThat(eventProcessor.executedPlans).hasSize(3);
+ applyLatestMaxInflightTasksChangedEvent(eventManager, manager, 1);
+
+ assertThat(eventProcessor.executedPlans).hasSize(4);
+ assertThat(eventProcessor.executedPlans.get(3).getTableBucket()).isEqualTo(buckets.get(3));
+ assertThat(countStatus(manager, RebalanceStatus.REBALANCING)).isEqualTo(1);
+
+ manager.close();
+ }
+
+ @Test
+ void testTimeoutEnqueuesEventsForAllInflightTasks() throws Exception {
+ ManualClock clock = new ManualClock(0L);
+ RecordingEventManager eventManager = new RecordingEventManager();
+ NoOpScheduledExecutor executor = new NoOpScheduledExecutor();
+ Configuration conf = new Configuration();
+ conf.set(ConfigOptions.COORDINATOR_REBALANCE_MAX_INFLIGHT_TASKS, 2);
+ RecordingCoordinatorEventProcessor eventProcessor =
+ buildRecordingCoordinatorEventProcessor(conf);
+ RebalanceManager manager =
+ new RebalanceManager(
+ eventProcessor, zookeeperClient, eventManager, clock, conf, executor);
+
+ Map<TableBucket, RebalancePlanForBucket> plan = createRebalancePlan(3);
+ List<TableBucket> buckets = new ArrayList<>(plan.keySet());
+ zookeeperClient.registerRebalanceTask(
+ new RebalanceTask("timeout-all-inflight-test", NOT_STARTED, plan));
+ manager.registerRebalance("timeout-all-inflight-test", plan, NOT_STARTED);
+ assertThat(eventProcessor.executedPlans).hasSize(2);
+
+ clock.advanceTime(Duration.ofMillis(130_000));
+ manager.checkTimeout();
+
+ assertThat(eventManager.events).hasSize(2);
+ assertThat(((RebalanceTaskTimeoutEvent) eventManager.events.get(0)).getTableBucket())
+ .isEqualTo(buckets.get(0));
+ assertThat(((RebalanceTaskTimeoutEvent) eventManager.events.get(1)).getTableBucket())
+ .isEqualTo(buckets.get(1));
Review Comment:
Agreed. Fixed the test to avoid depending on the timeout event order. It now asserts the timed-out buckets with containsExactlyInAnyOrder, matching the unordered iteration behavior in checkTimeout().
--
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] Add in-memory rebalance round limit [fluss]
Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on code in PR #3622:
URL: https://github.com/apache/fluss/pull/3622#discussion_r3548699070
##########
fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java:
##########
@@ -110,6 +112,8 @@ class DynamicServerConfig {
void register(ServerReconfigurable serverReconfigurable) {
serverReconfigures.put(serverReconfigurable.getClass(), serverReconfigurable);
+ serverReconfigurable.validate(currentConfig);
+ serverReconfigurable.reconfigure(currentConfig);
}
Review Comment:
Thanks for catching this. Fixed by serializing register() with the same write lock used by dynamic config updates. The method now validates and applies the current config before adding the reconfigurable to serverReconfigures, so a failed registration will not leave a stale entry behind.
--
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