You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "swuferhong (via GitHub)" <gi...@apache.org> on 2026/07/14 13:34:00 UTC
[PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
swuferhong opened a new pull request, #3654:
URL: https://github.com/apache/fluss/pull/3654
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink], [rust], [python], [c++], [elixir]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
- **Generative AI disclosure:** Indicate whether generative AI tools were used in authoring this PR. If yes, specify the tool below.
- [ ] No generative AI tools used
- [ ] Yes (please specify the tool below)
**(The sections below can be removed for hotfixes or typos)**
-->
<!--
Generated-by: [Tool Name and Version] following [the guidelines](https://github.com/apache/fluss/blob/main/AGENTS.md)
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #3653
<!-- What is the purpose of the change -->
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
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3597222897
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
Review Comment:
Shall we fix this as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3584317702
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
@@ -145,6 +155,165 @@ public void handleStateChange(
}
}
+ private void batchHandleControlledShutdown(
+ Set<TableBucket> tableBuckets,
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection) {
+ Map<TableBucket, LeaderAndIsr> currentLeaderAndIsrs;
+ long batchReadStartTimeMs = System.currentTimeMillis();
+ try {
+ currentLeaderAndIsrs = zooKeeperClient.getLeaderAndIsrs(tableBuckets);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to batch read LeaderAndIsr for {} buckets during controlled shutdown. Falling back to individual updates.",
+ tableBuckets.size(),
+ e);
+ for (TableBucket tableBucket : tableBuckets) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ return;
+ }
+ LOG.info(
+ "Batch read LeaderAndIsr for {} of {} controlled shutdown buckets in {} ms.",
+ currentLeaderAndIsrs.size(),
+ tableBuckets.size(),
+ System.currentTimeMillis() - batchReadStartTimeMs);
+
+ long electionStartTimeMs = System.currentTimeMillis();
+ Map<TableBucket, ElectionResult> electionResults = new HashMap<>();
+ Map<TableBucket, String> partitionNames = new HashMap<>();
+ Set<TableBucket> bucketsToRetryIndividually = new HashSet<>();
+ for (TableBucket tableBucket : tableBuckets) {
+ coordinatorContext.putBucketStateIfNotExists(
+ tableBucket, BucketState.NonExistentBucket);
+ if (!checkValidTableBucketStateChange(tableBucket, BucketState.OnlineBucket)) {
+ continue;
+ }
+
+ BucketState currentState = coordinatorContext.getBucketState(tableBucket);
+ if (currentState == BucketState.NewBucket) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ String partitionName = null;
+ if (tableBucket.getPartitionId() != null) {
+ partitionName = coordinatorContext.getPartitionName(tableBucket.getPartitionId());
+ if (partitionName == null) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ String.format(
+ "Can't find partition name for partition: %s.",
+ tableBucket.getBucket()));
Review Comment:
The failure reason says it can't find the partition name for a partition, but it formats the bucket id (`getBucket()`) rather than the partition id. This makes troubleshooting misleading for partitioned tables.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3594457573
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
@@ -145,6 +155,165 @@ public void handleStateChange(
}
}
+ private void batchHandleControlledShutdown(
+ Set<TableBucket> tableBuckets,
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection) {
+ Map<TableBucket, LeaderAndIsr> currentLeaderAndIsrs;
+ long batchReadStartTimeMs = System.currentTimeMillis();
+ try {
+ currentLeaderAndIsrs = zooKeeperClient.getLeaderAndIsrs(tableBuckets);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to batch read LeaderAndIsr for {} buckets during controlled shutdown. Falling back to individual updates.",
+ tableBuckets.size(),
+ e);
+ for (TableBucket tableBucket : tableBuckets) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ return;
+ }
+ LOG.info(
+ "Batch read LeaderAndIsr for {} of {} controlled shutdown buckets in {} ms.",
+ currentLeaderAndIsrs.size(),
+ tableBuckets.size(),
+ System.currentTimeMillis() - batchReadStartTimeMs);
+
+ long electionStartTimeMs = System.currentTimeMillis();
+ Map<TableBucket, ElectionResult> electionResults = new HashMap<>();
+ Map<TableBucket, String> partitionNames = new HashMap<>();
+ Set<TableBucket> bucketsToRetryIndividually = new HashSet<>();
+ for (TableBucket tableBucket : tableBuckets) {
+ coordinatorContext.putBucketStateIfNotExists(
+ tableBucket, BucketState.NonExistentBucket);
+ if (!checkValidTableBucketStateChange(tableBucket, BucketState.OnlineBucket)) {
+ continue;
+ }
+
+ BucketState currentState = coordinatorContext.getBucketState(tableBucket);
+ if (currentState == BucketState.NewBucket) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ String partitionName = null;
+ if (tableBucket.getPartitionId() != null) {
+ partitionName = coordinatorContext.getPartitionName(tableBucket.getPartitionId());
+ if (partitionName == null) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ String.format(
+ "Can't find partition name for partition: %s.",
+ tableBucket.getPartitionId()));
+ continue;
+ }
+ }
+
+ LeaderAndIsr leaderAndIsr = currentLeaderAndIsrs.get(tableBucket);
+ if (leaderAndIsr == null) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ Optional<ElectionResult> optionalElectionResult =
+ electNewLeaderForTableBucket(
+ tableBucket, leaderAndIsr, controlledShutdownLeaderElection);
+ if (!optionalElectionResult.isPresent()) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ "Elect result is empty.");
+ continue;
+ }
+ electionResults.put(tableBucket, optionalElectionResult.get());
+ partitionNames.put(tableBucket, partitionName);
+ }
+ LOG.info(
+ "Prepared {} controlled shutdown leader elections in {} ms; {} buckets require individual retry.",
+ electionResults.size(),
+ System.currentTimeMillis() - electionStartTimeMs,
+ bucketsToRetryIndividually.size());
+
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrsToUpdate = new HashMap<>();
+ electionResults.forEach(
+ (tableBucket, electionResult) ->
+ leaderAndIsrsToUpdate.put(tableBucket, electionResult.leaderAndIsr));
+ Set<TableBucket> updatedBuckets =
+ batchUpdateLeaderAndIsrWithFallback(leaderAndIsrsToUpdate);
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "LiebingYu (via GitHub)" <gi...@apache.org>.
LiebingYu commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3587005230
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
@@ -145,6 +155,165 @@ public void handleStateChange(
}
}
+ private void batchHandleControlledShutdown(
+ Set<TableBucket> tableBuckets,
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection) {
+ Map<TableBucket, LeaderAndIsr> currentLeaderAndIsrs;
+ long batchReadStartTimeMs = System.currentTimeMillis();
+ try {
+ currentLeaderAndIsrs = zooKeeperClient.getLeaderAndIsrs(tableBuckets);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to batch read LeaderAndIsr for {} buckets during controlled shutdown. Falling back to individual updates.",
+ tableBuckets.size(),
+ e);
+ for (TableBucket tableBucket : tableBuckets) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ return;
+ }
+ LOG.info(
+ "Batch read LeaderAndIsr for {} of {} controlled shutdown buckets in {} ms.",
+ currentLeaderAndIsrs.size(),
+ tableBuckets.size(),
+ System.currentTimeMillis() - batchReadStartTimeMs);
+
+ long electionStartTimeMs = System.currentTimeMillis();
+ Map<TableBucket, ElectionResult> electionResults = new HashMap<>();
+ Map<TableBucket, String> partitionNames = new HashMap<>();
+ Set<TableBucket> bucketsToRetryIndividually = new HashSet<>();
+ for (TableBucket tableBucket : tableBuckets) {
+ coordinatorContext.putBucketStateIfNotExists(
+ tableBucket, BucketState.NonExistentBucket);
+ if (!checkValidTableBucketStateChange(tableBucket, BucketState.OnlineBucket)) {
+ continue;
+ }
+
+ BucketState currentState = coordinatorContext.getBucketState(tableBucket);
+ if (currentState == BucketState.NewBucket) {
+ bucketsToRetryIndividually.add(tableBucket);
Review Comment:
nit: The current controlled shutdown tests in `TableBucketStateMachineTest` only exercise the batch path with buckets already in `OnlineBucket` state. Consider adding a test with a mixed set — some buckets in `OnlineBucket` and some in `NewBucket`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3597231559
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong merged PR #3654:
URL: https://github.com/apache/fluss/pull/3654
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3593888602
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
@@ -145,6 +155,165 @@ public void handleStateChange(
}
}
+ private void batchHandleControlledShutdown(
+ Set<TableBucket> tableBuckets,
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection) {
+ Map<TableBucket, LeaderAndIsr> currentLeaderAndIsrs;
+ long batchReadStartTimeMs = System.currentTimeMillis();
+ try {
+ currentLeaderAndIsrs = zooKeeperClient.getLeaderAndIsrs(tableBuckets);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to batch read LeaderAndIsr for {} buckets during controlled shutdown. Falling back to individual updates.",
+ tableBuckets.size(),
+ e);
+ for (TableBucket tableBucket : tableBuckets) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ return;
+ }
+ LOG.info(
+ "Batch read LeaderAndIsr for {} of {} controlled shutdown buckets in {} ms.",
+ currentLeaderAndIsrs.size(),
+ tableBuckets.size(),
+ System.currentTimeMillis() - batchReadStartTimeMs);
+
+ long electionStartTimeMs = System.currentTimeMillis();
+ Map<TableBucket, ElectionResult> electionResults = new HashMap<>();
+ Map<TableBucket, String> partitionNames = new HashMap<>();
+ Set<TableBucket> bucketsToRetryIndividually = new HashSet<>();
+ for (TableBucket tableBucket : tableBuckets) {
+ coordinatorContext.putBucketStateIfNotExists(
+ tableBucket, BucketState.NonExistentBucket);
+ if (!checkValidTableBucketStateChange(tableBucket, BucketState.OnlineBucket)) {
+ continue;
+ }
+
+ BucketState currentState = coordinatorContext.getBucketState(tableBucket);
+ if (currentState == BucketState.NewBucket) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ String partitionName = null;
+ if (tableBucket.getPartitionId() != null) {
+ partitionName = coordinatorContext.getPartitionName(tableBucket.getPartitionId());
+ if (partitionName == null) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ String.format(
+ "Can't find partition name for partition: %s.",
+ tableBucket.getPartitionId()));
+ continue;
+ }
+ }
+
+ LeaderAndIsr leaderAndIsr = currentLeaderAndIsrs.get(tableBucket);
+ if (leaderAndIsr == null) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ Optional<ElectionResult> optionalElectionResult =
+ electNewLeaderForTableBucket(
+ tableBucket, leaderAndIsr, controlledShutdownLeaderElection);
+ if (!optionalElectionResult.isPresent()) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ "Elect result is empty.");
+ continue;
+ }
+ electionResults.put(tableBucket, optionalElectionResult.get());
+ partitionNames.put(tableBucket, partitionName);
+ }
+ LOG.info(
+ "Prepared {} controlled shutdown leader elections in {} ms; {} buckets require individual retry.",
+ electionResults.size(),
+ System.currentTimeMillis() - electionStartTimeMs,
+ bucketsToRetryIndividually.size());
+
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrsToUpdate = new HashMap<>();
+ electionResults.forEach(
+ (tableBucket, electionResult) ->
+ leaderAndIsrsToUpdate.put(tableBucket, electionResult.leaderAndIsr));
+ Set<TableBucket> updatedBuckets =
+ batchUpdateLeaderAndIsrWithFallback(leaderAndIsrsToUpdate);
+
+ for (TableBucket tableBucket : updatedBuckets) {
+ ElectionResult electionResult = electionResults.get(tableBucket);
+ coordinatorContext.putBucketLeaderAndIsr(tableBucket, electionResult.leaderAndIsr);
+ doStateChange(tableBucket, BucketState.OnlineBucket);
+ coordinatorRequestBatch.addNotifyLeaderRequestForTabletServers(
+ new HashSet<>(electionResult.liveReplicas),
+ PhysicalTablePath.of(
+ coordinatorContext.getTablePathById(tableBucket.getTableId()),
+ partitionNames.get(tableBucket)),
+ tableBucket,
+ coordinatorContext.getAssignment(tableBucket),
+ electionResult.leaderAndIsr);
+ }
+
+ for (TableBucket tableBucket : bucketsToRetryIndividually) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ }
+
+ private Set<TableBucket> batchUpdateLeaderAndIsrWithFallback(
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrs) {
+ Set<TableBucket> updatedBuckets = new HashSet<>();
+ if (leaderAndIsrs.isEmpty()) {
+ return updatedBuckets;
+ }
+
+ try {
+ zooKeeperClient.batchUpdateLeaderAndIsr(
Review 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?
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
@@ -145,6 +155,165 @@ public void handleStateChange(
}
}
+ private void batchHandleControlledShutdown(
+ Set<TableBucket> tableBuckets,
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection) {
+ Map<TableBucket, LeaderAndIsr> currentLeaderAndIsrs;
+ long batchReadStartTimeMs = System.currentTimeMillis();
+ try {
+ currentLeaderAndIsrs = zooKeeperClient.getLeaderAndIsrs(tableBuckets);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to batch read LeaderAndIsr for {} buckets during controlled shutdown. Falling back to individual updates.",
+ tableBuckets.size(),
+ e);
+ for (TableBucket tableBucket : tableBuckets) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ return;
+ }
+ LOG.info(
+ "Batch read LeaderAndIsr for {} of {} controlled shutdown buckets in {} ms.",
+ currentLeaderAndIsrs.size(),
+ tableBuckets.size(),
+ System.currentTimeMillis() - batchReadStartTimeMs);
+
+ long electionStartTimeMs = System.currentTimeMillis();
+ Map<TableBucket, ElectionResult> electionResults = new HashMap<>();
+ Map<TableBucket, String> partitionNames = new HashMap<>();
+ Set<TableBucket> bucketsToRetryIndividually = new HashSet<>();
+ for (TableBucket tableBucket : tableBuckets) {
+ coordinatorContext.putBucketStateIfNotExists(
+ tableBucket, BucketState.NonExistentBucket);
+ if (!checkValidTableBucketStateChange(tableBucket, BucketState.OnlineBucket)) {
+ continue;
+ }
+
+ BucketState currentState = coordinatorContext.getBucketState(tableBucket);
+ if (currentState == BucketState.NewBucket) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ String partitionName = null;
+ if (tableBucket.getPartitionId() != null) {
+ partitionName = coordinatorContext.getPartitionName(tableBucket.getPartitionId());
+ if (partitionName == null) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ String.format(
+ "Can't find partition name for partition: %s.",
+ tableBucket.getPartitionId()));
+ continue;
+ }
+ }
+
+ LeaderAndIsr leaderAndIsr = currentLeaderAndIsrs.get(tableBucket);
+ if (leaderAndIsr == null) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ Optional<ElectionResult> optionalElectionResult =
+ electNewLeaderForTableBucket(
+ tableBucket, leaderAndIsr, controlledShutdownLeaderElection);
+ if (!optionalElectionResult.isPresent()) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ "Elect result is empty.");
+ continue;
+ }
+ electionResults.put(tableBucket, optionalElectionResult.get());
+ partitionNames.put(tableBucket, partitionName);
+ }
+ LOG.info(
+ "Prepared {} controlled shutdown leader elections in {} ms; {} buckets require individual retry.",
+ electionResults.size(),
+ System.currentTimeMillis() - electionStartTimeMs,
+ bucketsToRetryIndividually.size());
+
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrsToUpdate = new HashMap<>();
+ electionResults.forEach(
+ (tableBucket, electionResult) ->
+ leaderAndIsrsToUpdate.put(tableBucket, electionResult.leaderAndIsr));
+ Set<TableBucket> updatedBuckets =
+ batchUpdateLeaderAndIsrWithFallback(leaderAndIsrsToUpdate);
Review Comment:
If BadVersionException occurs, need to stop here directly, on need to bucketsToRetryIndividually later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [server] Batch LeaderAndIsr updates for controlled shutdown [fluss]
Posted by "swuferhong (via GitHub)" <gi...@apache.org>.
swuferhong commented on code in PR #3654:
URL: https://github.com/apache/fluss/pull/3654#discussion_r3594428360
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java:
##########
@@ -145,6 +155,165 @@ public void handleStateChange(
}
}
+ private void batchHandleControlledShutdown(
+ Set<TableBucket> tableBuckets,
+ ControlledShutdownLeaderElection controlledShutdownLeaderElection) {
+ Map<TableBucket, LeaderAndIsr> currentLeaderAndIsrs;
+ long batchReadStartTimeMs = System.currentTimeMillis();
+ try {
+ currentLeaderAndIsrs = zooKeeperClient.getLeaderAndIsrs(tableBuckets);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to batch read LeaderAndIsr for {} buckets during controlled shutdown. Falling back to individual updates.",
+ tableBuckets.size(),
+ e);
+ for (TableBucket tableBucket : tableBuckets) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ return;
+ }
+ LOG.info(
+ "Batch read LeaderAndIsr for {} of {} controlled shutdown buckets in {} ms.",
+ currentLeaderAndIsrs.size(),
+ tableBuckets.size(),
+ System.currentTimeMillis() - batchReadStartTimeMs);
+
+ long electionStartTimeMs = System.currentTimeMillis();
+ Map<TableBucket, ElectionResult> electionResults = new HashMap<>();
+ Map<TableBucket, String> partitionNames = new HashMap<>();
+ Set<TableBucket> bucketsToRetryIndividually = new HashSet<>();
+ for (TableBucket tableBucket : tableBuckets) {
+ coordinatorContext.putBucketStateIfNotExists(
+ tableBucket, BucketState.NonExistentBucket);
+ if (!checkValidTableBucketStateChange(tableBucket, BucketState.OnlineBucket)) {
+ continue;
+ }
+
+ BucketState currentState = coordinatorContext.getBucketState(tableBucket);
+ if (currentState == BucketState.NewBucket) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ String partitionName = null;
+ if (tableBucket.getPartitionId() != null) {
+ partitionName = coordinatorContext.getPartitionName(tableBucket.getPartitionId());
+ if (partitionName == null) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ String.format(
+ "Can't find partition name for partition: %s.",
+ tableBucket.getPartitionId()));
+ continue;
+ }
+ }
+
+ LeaderAndIsr leaderAndIsr = currentLeaderAndIsrs.get(tableBucket);
+ if (leaderAndIsr == null) {
+ bucketsToRetryIndividually.add(tableBucket);
+ continue;
+ }
+
+ Optional<ElectionResult> optionalElectionResult =
+ electNewLeaderForTableBucket(
+ tableBucket, leaderAndIsr, controlledShutdownLeaderElection);
+ if (!optionalElectionResult.isPresent()) {
+ logFailedStateChange(
+ tableBucket,
+ currentState,
+ BucketState.OnlineBucket,
+ "Elect result is empty.");
+ continue;
+ }
+ electionResults.put(tableBucket, optionalElectionResult.get());
+ partitionNames.put(tableBucket, partitionName);
+ }
+ LOG.info(
+ "Prepared {} controlled shutdown leader elections in {} ms; {} buckets require individual retry.",
+ electionResults.size(),
+ System.currentTimeMillis() - electionStartTimeMs,
+ bucketsToRetryIndividually.size());
+
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrsToUpdate = new HashMap<>();
+ electionResults.forEach(
+ (tableBucket, electionResult) ->
+ leaderAndIsrsToUpdate.put(tableBucket, electionResult.leaderAndIsr));
+ Set<TableBucket> updatedBuckets =
+ batchUpdateLeaderAndIsrWithFallback(leaderAndIsrsToUpdate);
+
+ for (TableBucket tableBucket : updatedBuckets) {
+ ElectionResult electionResult = electionResults.get(tableBucket);
+ coordinatorContext.putBucketLeaderAndIsr(tableBucket, electionResult.leaderAndIsr);
+ doStateChange(tableBucket, BucketState.OnlineBucket);
+ coordinatorRequestBatch.addNotifyLeaderRequestForTabletServers(
+ new HashSet<>(electionResult.liveReplicas),
+ PhysicalTablePath.of(
+ coordinatorContext.getTablePathById(tableBucket.getTableId()),
+ partitionNames.get(tableBucket)),
+ tableBucket,
+ coordinatorContext.getAssignment(tableBucket),
+ electionResult.leaderAndIsr);
+ }
+
+ for (TableBucket tableBucket : bucketsToRetryIndividually) {
+ doHandleStateChange(
+ tableBucket, BucketState.OnlineBucket, controlledShutdownLeaderElection);
+ }
+ }
+
+ private Set<TableBucket> batchUpdateLeaderAndIsrWithFallback(
+ Map<TableBucket, LeaderAndIsr> leaderAndIsrs) {
+ Set<TableBucket> updatedBuckets = new HashSet<>();
+ if (leaderAndIsrs.isEmpty()) {
+ return updatedBuckets;
+ }
+
+ try {
+ zooKeeperClient.batchUpdateLeaderAndIsr(
Review 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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org