[server] Add listRebalances API and expose rebalance timestamps - #3985
[server] Add listRebalances API and expose rebalance timestamps#3985morazow wants to merge 2 commits into
Conversation
5b2283c to
64191c5
Compare
8569c10 to
014a5ac
Compare
|
Scope note for reviewers: per-bucket detail for historical rebalances This PR keeps But It is easier to enable this API: Keeping it out of this PR keeps the review surface small and leaves one design point open for discussion:
The follow-up also changes one contract: |
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.
014a5ac to
8757b6a
Compare
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