Skip to content

[server] Add listRebalances API and expose rebalance timestamps - #3985

Open
morazow wants to merge 2 commits into
apache:mainfrom
morazow:rebalance-progress
Open

[server] Add listRebalances API and expose rebalance timestamps#3985
morazow wants to merge 2 commits into
apache:mainfrom
morazow:rebalance-progress

Conversation

@morazow

@morazow morazow commented Aug 13, 2026

Copy link
Copy Markdown
Contributor

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

  • RebalanceTask carries startedAtMs/completedAtMs (JSON serde v2; v1 znodes read
    back as -1, meaning unset).
  • On completion or cancellation the final task is also written to a bounded ZooKeeper
    history at /cluster/rebalance_history/<rebalanceId> (last 10 kept). A failed history
    write is logged and never fails the transition; corrupt entries are skipped, not fatal.
  • New Admin#listRebalances() (ApiKeys.LIST_REBALANCES) returns a RebalanceInfo
    summary - 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

  • RebalanceProgress gains startedAtMs()/completedAtMs(), transported by two new
    optional fields on ListRebalanceProgressResponse (-1 maps to absent).
  • RebalanceProgressJsonSerializer emits both when set.

SQL surface

  • sys.list_rebalance gains started_at/completed_at TIMESTAMP_LTZ(3) columns.
  • Called without a rebalance id it now returns one row per known rebalance (current plus
    retained history, newest first) instead of only the current one. Historical rows carry
    null progress and plan detail.
  • Docs updated: engine-flink/procedures.md, maintenance/operations/rebalance.md.

Closes #3965.

Notes for reviewers

Where the ZooKeeper read happens. listRebalances needs both in-memory state and
ZooKeeper history. CoordinatorEventProcessor#processListRebalances snapshots the current
rebalance on the coordinator event thread, where it mutates, then reads the history on the
ioExecutor so the event loop never blocks on ZooKeeper.
RebalanceManager#listRebalances(RebalanceInfo) takes that snapshot as a parameter and
touches nothing else, which is what makes it safe off the event thread.

The no-id procedure path issues two RPCs - listRebalanceProgress(null) for the
current 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

  • Unit: 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 restore
    with and without timestamps, no re-completion of an already-final restored task,
    newest-first ordering, empty history), CoordinatorEventProcessorTest (the response
    callback completes when the rebalance manager is closed), ClientRpcMessageUtilsTest
    (-1 <-> absent mapping), RebalanceProgressJsonSerializerTest.
  • Integration: RebalanceITCase (end-to-end listRebalances and progress timestamps on a
    live cluster), FlussAuthorizationITCase#testListRebalances (DESCRIBE on
    Resource.cluster(), deny and allow), FlinkProcedureITCase (both procedure shapes).

🤖 AI-assisted changes - reviewed by human developer

@morazow
morazow force-pushed the rebalance-progress branch from 5b2283c to 64191c5 Compare August 18, 2026 13:49
@morazow morazow changed the title [server] Retain bounded rebalance history and add listRebalances API [server] Add listRebalances API and expose rebalance timestamps Aug 18, 2026
@morazow
morazow force-pushed the rebalance-progress branch 5 times, most recently from 8569c10 to 014a5ac Compare August 19, 2026 08:44
@morazow

morazow commented Aug 19, 2026

Copy link
Copy Markdown
Contributor Author

Scope note for reviewers: per-bucket detail for historical rebalances

This PR keeps listRebalances() deliberately summary-only (id, status, timestamps). The per-bucket plan of a retained historical rebalance is already persisted, each history znode stores the full RebalanceTask, plan included.

But listRebalanceProgress(id) still throws NoRebalanceInProgressException for non-current ids, so that plan is not yet reachable through the API.

It is easier to enable this API: listRebalanceProgress(id) fall back to the ZooKeeper history when the id is not current, and build the RebalanceProgress from the stored plan. No proto or serde change is needed; the existing response message already carries per-bucket plans.

Keeping it out of this PR keeps the review surface small and leaves one design point open for discussion:

  • History stores plans plus the overall final status, not per-bucket outcomes (those live only in coordinator memory). A historical RebalanceProgress would stamp every bucket with the overall status — exact for COMPLETED, approximate for FAILED/CANCELED. If per-bucket outcomes matter, the follow-up would instead need a serde bump to store RebalanceResultForBucket per bucket. Opinions welcome.

The follow-up also changes one contract: NoRebalanceInProgressException would no longer be thrown for ids still in Zookeeper, which I read as the intent of #3965.

@morazow
morazow marked this pull request as ready for review August 19, 2026 09:30
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
morazow force-pushed the rebalance-progress branch from 014a5ac to 8757b6a Compare August 19, 2026 22:02
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add completion timestamp and historical listing to RebalanceProgress

1 participant