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