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