[server] Add in-memory rebalance round limit#3622
Conversation
|
This PR is derived from the previous rebalance throttling work in #3498 and #3537, please help review, thanks! @swuferhong @wuchong |
There was a problem hiding this comment.
Pull request overview
Adds coordinator-side rebalance throttling that can (a) cap concurrent bucket movements dynamically and (b) optionally admit bucket tasks in bounded in-memory “rounds” to avoid activating very large plans all at once, without introducing new ZooKeeper metadata models.
Changes:
- Introduce
coordinator.rebalance.max-buckets-per-round(static, default0) and implement in-memory round admission inRebalanceManagerwith deterministic bucket ordering. - Wire
coordinator.rebalance.max-inflight-tasksinto the dynamic-config framework for coordinators via coordinator events, plus recovery via a dedicatedRecoverRebalanceEvent. - Update docs and expand tests for dynamic config changes, recovery, inflight throttling, and round behavior.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| website/docs/maintenance/operations/updating-configs.md | Document coordinator.rebalance.max-inflight-tasks as a dynamically updatable option. |
| website/docs/maintenance/operations/rebalance.md | Add operational guidance for tuning inflight concurrency and per-round admission. |
| website/docs/maintenance/configuration.md | Clarify restart vs dynamic config behavior; document new rebalance options. |
| fluss-server/src/test/java/org/apache/fluss/server/DynamicConfigChangeTest.java | Add tests validating dynamic updates for coordinator.rebalance.max-inflight-tasks. |
| fluss-server/src/test/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManagerTest.java | Add/adjust tests for inflight throttling, round admission, recovery, and timeouts. |
| fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java | Update in-progress rebalance counting logic for new manager behavior. |
| fluss-server/src/main/java/org/apache/fluss/server/DynamicServerConfig.java | Allow dynamic updates for max inflight tasks; change registration behavior for reconfigurables. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java | Implement in-memory round/window scheduling and dynamic max-inflight updates via events. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/RecoverRebalanceEvent.java | New event to recover persisted rebalance tasks on the coordinator event thread. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/event/RebalanceMaxInflightTasksChangedEvent.java | New event to apply dynamic max-inflight changes on the coordinator event thread. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorServer.java | Register RebalanceManager with DynamicConfigManager. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java | Handle new recovery/config-change events and pass config into RebalanceManager. |
| fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java | Add config definitions for rebalance inflight and per-round limits. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| void register(ServerReconfigurable serverReconfigurable) { | ||
| serverReconfigures.put(serverReconfigurable.getClass(), serverReconfigurable); | ||
| serverReconfigurable.validate(currentConfig); | ||
| serverReconfigurable.reconfigure(currentConfig); | ||
| } |
There was a problem hiding this 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.
| RebalanceStatus status = | ||
| eventProcessor | ||
| .getRebalanceManager() | ||
| .listRebalanceProgress(null) | ||
| .progressForBucketMap() | ||
| .get(tb) | ||
| .status(); | ||
| if (!RebalanceStatus.FINAL_STATUSES.contains(status)) { | ||
| count++; | ||
| } |
There was a problem hiding this 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.
| 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)); |
There was a problem hiding this 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().
Purpose
Linked issue: #3348 #3536
This PR is derived from the previous rebalance throttling work in #3498 and #3537.
#3537 adds
coordinator.rebalance.max-inflight-tasksto 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
RebalanceManagerand does not add any new ZooKeeper model, serde, or znode.The default behavior remains unchanged.
Brief change log
Add
coordinator.rebalance.max-buckets-per-roundwith default value0.0disables round limiting.Extend
RebalanceManagerwith an in-memory rebalance round/window.RebalanceTaskznode.tableId,partitionId, andbucketId.coordinator.rebalance.max-buckets-per-roundbucket tasks into the current round.coordinator.rebalance.max-inflight-tasksfrom [server] Support configurable rebalance concurrency #3537 to control concurrent execution within the active round.Keep the implementation ZooKeeper-compatible with the existing rebalance flow.
RebalanceTaskand rebuilds the in-memory scheduling window after CoordinatorServer restart.Update rebalance documentation to explain how:
coordinator.rebalance.max-inflight-taskscontrols concurrent bucket movements.coordinator.rebalance.max-buckets-per-roundcontrols per-round admission.1, can create many small scheduling rounds and should be used only when strict serial admission is required.Tests
git diff --checkmvn -pl fluss-server -am -DskipITs -Dcheckstyle.skip=true -DfailIfNoTests=false -Dtest=RebalanceManagerTest testAPI and Format
Adds a new coordinator configuration:
coordinator.rebalance.max-buckets-per-roundNo public client API changes.
No RPC protocol changes.
No storage format changes.
No new ZooKeeper metadata, serde, or znode.
Backward compatibility:
0preserves the existing behavior.coordinator.rebalance.max-inflight-taskskeeps the [server] Support configurable rebalance concurrency #3537 behavior and remains responsible for concurrent execution throttling.Documentation
coordinator.rebalance.max-buckets-per-roundtowebsite/docs/maintenance/configuration.md.website/docs/maintenance/operations/rebalance.mdwith operational guidance for usingcoordinator.rebalance.max-inflight-tasksandcoordinator.rebalance.max-buckets- per-roundtogether.