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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
