[server] Add listRebalances API and expose rebalance timestamps - #3985
Draft
morazow wants to merge 2 commits into
Draft
[server] Add listRebalances API and expose rebalance timestamps#3985morazow wants to merge 2 commits into
morazow wants to merge 2 commits into
Conversation
Rebalance tasks now carry started/completed timestamps (RebalanceTask JSON serde v2, backward compatible with v1 znodes). On completion or cancellation the final task is also written to a bounded ZooKeeper history at /cluster/rebalance_history/<rebalanceId> (last 10 entries kept), and the new Admin#listRebalances() API returns a summary (id, status, timestamps) of the current rebalance plus the retained history, newest first. Per-bucket detail remains available via listRebalanceProgress(id). Part of apache#3965 (timestamps on listRebalanceProgress follow separately).
RebalanceProgress now carries startedAtMs/completedAtMs (epoch millis, -1 when unset), populated by the coordinator and transported via two new optional fields on ListRebalanceProgressResponse. The progress JSON serializer includes the timestamps when set. The sys.list_rebalance() Flink procedure gains started_at/completed_at TIMESTAMP_LTZ(3) output columns, and when called without a rebalance id it now returns one row per known rebalance (the current one plus the retained history via listRebalances(), newest first) instead of only the current one; historical rows carry null progress and plan detail. Closes apache#3965.
morazow
force-pushed
the
rebalance-progress
branch
from
August 18, 2026 13:49
5b2283c to
64191c5
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Rebalance operations were observable only while in flight:
listRebalanceProgress(id)served the current rebalance, and nothing recorded when one started or finished. This PR
adds the history and the timestamps, then surfaces them through the SQL procedure.
Storage and the new listing API
RebalanceTaskcarriesstartedAtMs/completedAtMs(JSON serde v2; v1 znodes readback as
-1, meaning unset).history at
/cluster/rebalance_history/<rebalanceId>(last 10 kept). A failed historywrite is logged and never fails the transition; corrupt entries are skipped, not fatal.
Admin#listRebalances()(ApiKeys.LIST_REBALANCES) returns aRebalanceInfosummary - id, status, timestamps - for the current rebalance plus the retained history,
newest first. Per-bucket detail stays on
listRebalanceProgress(id).Timestamps on the existing progress RPC
RebalanceProgressgainsstartedAtMs()/completedAtMs(), transported by two newoptional fields on
ListRebalanceProgressResponse(-1maps to absent).RebalanceProgressJsonSerializeremits both when set.SQL surface
sys.list_rebalancegainsstarted_at/completed_atTIMESTAMP_LTZ(3)columns.retained history, newest first) instead of only the current one. Historical rows carry
null progress and plan detail.
engine-flink/procedures.md,maintenance/operations/rebalance.md.Closes #3965.
Notes for reviewers
Where the ZooKeeper read happens.
listRebalancesneeds both in-memory state andZooKeeper history.
CoordinatorEventProcessor#processListRebalancessnapshots the currentrebalance on the coordinator event thread, where it mutates, then reads the history on the
ioExecutorso the event loop never blocks on ZooKeeper.RebalanceManager#listRebalances(RebalanceInfo)takes that snapshot as a parameter andtouches nothing else, which is what makes it safe off the event thread.
The no-id procedure path issues two RPCs -
listRebalanceProgress(null)for thecurrent entry's detail plus
listRebalances()for the summaries, stitched client-side -so a rebalance finishing between them can render one transiently stale row, corrected on
the next call. The alternative is carrying progress and plan detail for the current entry
inside
ListRebalancesResponse; that was left out to keep the new response summary-only.Happy to switch to the single-snapshot shape if you prefer it.
Two commits. The first adds the storage groundwork and the listing API, the second the
timestamp exposure and the SQL surface. Each compiles and passes its tests on its own, so
the PR is fine to rebase-merge or squash.
Test Plan
RebalanceTaskJsonSerdeTest(serde v2 round-trip, v1 document reads back as-1),ZooKeeperClientTest(history retention, idempotent re-register, corrupt-entry skip,sibling-znode isolation),
RebalanceManagerTest(timestamp stamping, failover restorewith and without timestamps, no re-completion of an already-final restored task,
newest-first ordering, empty history),
CoordinatorEventProcessorTest(the responsecallback completes when the rebalance manager is closed),
ClientRpcMessageUtilsTest(
-1<-> absent mapping),RebalanceProgressJsonSerializerTest.RebalanceITCase(end-to-endlistRebalancesand progress timestamps on alive cluster),
FlussAuthorizationITCase#testListRebalances(DESCRIBEonResource.cluster(), deny and allow),FlinkProcedureITCase(both procedure shapes).🤖 AI-assisted changes - reviewed by human developer