[server] Batch LeaderAndIsr updates for controlled shutdown#3654
[server] Batch LeaderAndIsr updates for controlled shutdown#3654swuferhong wants to merge 4 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR optimizes controlled shutdown leader migration in the coordinator by batching ZooKeeper LeaderAndIsr reads and updates for buckets led by a shutting-down tablet server, with fallbacks to the existing per-bucket path when batch reads are incomplete or batch updates fail. It also adds timing logs to better observe controlled shutdown performance, and expands unit tests to validate the new batching + fallback behavior.
Changes:
- Add a controlled-shutdown-specific batching path in
TableBucketStateMachinethat batch-readsLeaderAndIsr, prepares elections, and batch-updates ZooKeeper with safe fallbacks. - Improve ZooKeeper batch update logging (reduce per-bucket log verbosity; add aggregate timing/transaction count).
- Add/extend tests to cover batch behavior and fallback scenarios; add controlled shutdown timing logs on coordinator and tablet server.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java | Implements batch leader election + batch ZK update for controlled shutdown, with fallback to individual operations. |
| fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java | Adds timing/transaction-count logging for batchUpdateLeaderAndIsr and reduces per-bucket logs to debug. |
| fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java | Adds timing logs around controlled shutdown processing and avoids repeated set lookups. |
| fluss-server/src/main/java/org/apache/fluss/server/tablet/TabletServer.java | Adds duration/success logging for tablet server controlled shutdown attempts. |
| fluss-server/src/test/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachineTest.java | Expands controlled shutdown tests to assert batch read/update behavior and fallback paths. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
swuferhong
force-pushed
the
controll-shutdonw-opt-1
branch
from
c2302d2 to
d7505d1
Compare
LiebingYu
left a comment
LiebingYu
left a comment
There was a problem hiding this comment.
@swuferhong Thansk for the PR. Only minor comments.
loserwang1024
left a comment
loserwang1024
left a comment
There was a problem hiding this comment.
I left some minor comment
| } | ||
|
|
||
| try { | ||
| zooKeeperClient.batchUpdateLeaderAndIsr( |
There was a problem hiding this comment.
batchUpdateLeaderAndIsr() splits the input into multiple ZooKeeper transactions of at most MAX_BATCH_SIZE, so an exception may be thrown after some earlier transactions have already committed. The current fallback retries every bucket individually, including buckets from transactions that were already confirmed successful.
Rewriting them should still succeed because the version check is against the coordinator epoch znode, which is not changed by LeaderAndIsr updates. Therefore, this appears correct but results in redundant ZooKeeper writes. Would it be worth propagating the set of definitely committed buckets from batchUpdateLeaderAndIsr() so that only failed or uncertain transactions need to be retried?
There was a problem hiding this comment.
I think these retries write the same LeaderAndIsr values and validate against the coordinator epoch znode rather than the LeaderAndIsr node version, so they are idempotent and do not cause a correctness issue. The downside is limited to redundant ZooKeeper writes on the exceptional path.
Propagating the definitely committed buckets would require changing the ZooKeeper client API or introducing a partial-progress exception/result type. Given the additional complexity and the goal of keeping this PR minimally invasive, I suggest leaving it unchanged for now and considering it as a separate follow-up optimization.
| (tableBucket, electionResult) -> | ||
| leaderAndIsrsToUpdate.put(tableBucket, electionResult.leaderAndIsr)); | ||
| Set<TableBucket> updatedBuckets = | ||
| batchUpdateLeaderAndIsrWithFallback(leaderAndIsrsToUpdate); |
There was a problem hiding this comment.
If BadVersionException occurs, need to stop here directly, on need to bucketsToRetryIndividually later.
There was a problem hiding this comment.
A simple return may be unsafe because earlier ZooKeeper transactions or individual retries may have already succeeded. Returning immediately would skip updating the coordinator cache for those buckets. Since subsequent fenced writes will fail safely, I prefer keeping the current behavior.
There was a problem hiding this comment.
@swuferhong Thank you, LGTM 👍
UPD:
a couple of inconsistencies, listed them, but other than this - all good.
| BucketState.OnlineBucket, | ||
| String.format( | ||
| "Can't find partition name for partition: %s.", | ||
| tableBucket.getBucket())); |
There was a problem hiding this comment.
Shall we fix this as well?
| targetState, | ||
| String.format( | ||
| "Can't find partition name for partition: %s.", | ||
| tableBucket.getBucket())); |
Purpose
Linked issue: close #3653
Optimize controlled shutdown leader migration by batching LeaderAndIsr reads and updates in ZooKeeper. The new path is only used for controlled shutdown leader election and falls back to individual updates when batch reads are incomplete or batch updates fail.
This reduces the coordinator-side cost when a shutting-down tablet server owns many leader buckets, while preserving the existing election semantics.
Brief change log
Tests
API and Format
Documentation