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