Copilot commented on code in PR #3498:
URL: https://github.com/apache/fluss/pull/3498#discussion_r3441717012
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java:
##########
@@ -366,6 +436,303 @@ public RebalanceTask generateRebalanceTask(List<Goal>
goalsByPriority) {
return buildRebalanceTask(rebalanceId, rebalancePlanForBuckets);
}
+ /**
+ * Generates and registers a rebalance task. Large plans are split into
recoverable rounds when
+ * {@link ConfigOptions#COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND} is
configured with a
+ * positive value.
+ */
+ public RebalanceTask generateAndRegisterRebalance(List<Goal>
goalsByPriority) {
+ checkNotClosed();
+ RebalanceTask rebalanceTask = generateRebalanceTask(goalsByPriority);
+ try {
+ return registerGeneratedRebalanceTask(rebalanceTask);
+ } catch (Exception e) {
+ throw new RebalanceFailureException(
+ String.format(
+ "Failed to generate plan and execute rebalance.
The root cause: %s",
+ e.getMessage()),
+ e);
+ }
+ }
+
+ @VisibleForTesting
+ RebalanceTask registerGeneratedRebalanceTask(RebalanceTask rebalanceTask)
throws Exception {
+ Map<TableBucket, RebalancePlanForBucket> executePlan =
rebalanceTask.getExecutePlan();
+
+ zkClient.deleteRebalanceTask();
+ if (!shouldSplitRebalancePlan(executePlan)) {
+ currentRebalanceExecution = null;
+ zkClient.registerRebalanceTask(rebalanceTask);
+ registerRebalance(
+ rebalanceTask.getRebalanceId(), executePlan,
RebalanceStatus.NOT_STARTED);
+ return rebalanceTask;
+ }
+
+ List<RebalanceRound> rebalanceRounds = splitRebalancePlan(executePlan);
+ RebalanceExecution rebalanceExecution =
+ new RebalanceExecution(
+ rebalanceTask.getRebalanceId(),
+ REBALANCING,
+ rebalanceMaxBucketsPerRound,
+ 0,
+ rebalanceRounds.size());
+
+ for (RebalanceRound rebalanceRound : rebalanceRounds) {
+ zkClient.registerRebalanceRound(rebalanceRound);
+ }
+ zkClient.registerRebalanceExecution(rebalanceExecution);
+
+ currentRebalanceExecution = rebalanceExecution;
+ RebalanceRound firstRound = rebalanceRounds.get(0);
+ RebalanceTask firstRoundTask =
+ new RebalanceTask(
+ rebalanceTask.getRebalanceId(),
+ RebalanceStatus.NOT_STARTED,
+ firstRound.getExecutePlan());
+ zkClient.registerRebalanceTask(firstRoundTask);
+ registerRebalanceResults(
+ rebalanceTask.getRebalanceId(),
+ firstRound.getProgressForBucketMap(),
+ RebalanceStatus.NOT_STARTED);
+ LOG.info(
+ "Split rebalance task {} into {} rounds by {}={}, current
round buckets {}.",
+ rebalanceTask.getRebalanceId(),
+ rebalanceRounds.size(),
+
ConfigOptions.COORDINATOR_REBALANCE_MAX_BUCKETS_PER_ROUND.key(),
+ rebalanceMaxBucketsPerRound,
+ firstRound.getProgressForBucketMap().size());
+ return firstRoundTask;
+ }
+
+ private boolean shouldSplitRebalancePlan(Map<TableBucket,
RebalancePlanForBucket> executePlan) {
+ return rebalanceMaxBucketsPerRound > 0 && executePlan.size() >
rebalanceMaxBucketsPerRound;
+ }
+
+ private List<RebalanceRound> splitRebalancePlan(
+ Map<TableBucket, RebalancePlanForBucket> executePlan) {
+ List<Map.Entry<TableBucket, RebalancePlanForBucket>> sortedEntries =
+ new ArrayList<>(executePlan.entrySet());
+ sortedEntries.sort(
+ Comparator.comparing(
+ (Map.Entry<TableBucket,
RebalancePlanForBucket> entry) ->
+ entry.getKey().getTableId())
+ .thenComparing(
+ entry ->
+ entry.getKey().getPartitionId() == null
+ ? Long.MIN_VALUE
+ :
entry.getKey().getPartitionId())
+ .thenComparing(entry -> entry.getKey().getBucket()));
+
+ List<RebalanceRound> rebalanceRounds = new ArrayList<>();
+ for (int i = 0; i < sortedEntries.size(); i +=
rebalanceMaxBucketsPerRound) {
+ int end = Math.min(i + rebalanceMaxBucketsPerRound,
sortedEntries.size());
+ Map<TableBucket, RebalancePlanForBucket> roundPlan = new
LinkedHashMap<>();
+ for (int j = i; j < end; j++) {
+ Map.Entry<TableBucket, RebalancePlanForBucket> entry =
sortedEntries.get(j);
+ roundPlan.put(entry.getKey(), entry.getValue());
+ }
+ rebalanceRounds.add(RebalanceRound.ofPlan(rebalanceRounds.size(),
roundPlan));
+ }
+ return rebalanceRounds;
+ }
+
+ private void restoreRoundBasedRebalance(RebalanceExecution
rebalanceExecution)
+ throws Exception {
+ List<RebalanceRound> rebalanceRounds = zkClient.getRebalanceRounds();
+ currentRebalanceId = rebalanceExecution.getRebalanceId();
+ rebalanceStatus = rebalanceExecution.getRebalanceStatus();
+ currentRebalanceExecution = rebalanceExecution;
+
+ if (FINAL_STATUSES.contains(rebalanceExecution.getRebalanceStatus())) {
+ LOG.info(
+ "Restore final round-based rebalance {} with status {}.",
+ rebalanceExecution.getRebalanceId(),
+ rebalanceExecution.getRebalanceStatus());
+ return;
+ }
+
+ Optional<RebalanceRound> roundToResume =
findFirstUnfinishedRound(rebalanceRounds);
+ if (!roundToResume.isPresent()) {
+ RebalanceExecution completedExecution =
rebalanceExecution.withStatus(COMPLETED);
+ zkClient.registerRebalanceExecution(completedExecution);
+ currentRebalanceExecution = completedExecution;
+ rebalanceStatus = COMPLETED;
+ return;
+ }
+
+ RebalanceRound rebalanceRound = roundToResume.get();
+ RebalanceExecution resumedExecution =
+ rebalanceExecution.withStatusAndCurrentRound(
+ REBALANCING, rebalanceRound.getRoundIndex());
+ zkClient.registerRebalanceExecution(resumedExecution);
+ currentRebalanceExecution = resumedExecution;
+ zkClient.registerRebalanceTask(
+ new RebalanceTask(
+ resumedExecution.getRebalanceId(),
+ RebalanceStatus.NOT_STARTED,
+ rebalanceRound.getExecutePlan()));
+ registerRebalanceResults(
+ resumedExecution.getRebalanceId(),
+ rebalanceRound.getProgressForBucketMap(),
+ RebalanceStatus.NOT_STARTED);
+ }
+
+ private Optional<RebalanceRound>
findFirstUnfinishedRound(List<RebalanceRound> rounds) {
+ for (RebalanceRound round : rounds) {
+ if (round.hasUnfinishedTasks()) {
+ return Optional.of(round);
+ }
+ }
+ return Optional.empty();
+ }
+
+ private void updateCurrentRoundBucketStatus(
+ TableBucket tableBucket,
+ RebalancePlanForBucket planForBucket,
+ RebalanceStatus rebalanceStatus) {
+ RebalanceExecution rebalanceExecution = currentRebalanceExecution;
+ if (rebalanceExecution == null) {
+ return;
+ }
+
+ try {
+ Optional<RebalanceRound> rebalanceRound =
+
zkClient.getRebalanceRound(rebalanceExecution.getCurrentRound());
+ if (!rebalanceRound.isPresent()) {
+ LOG.warn(
+ "No rebalance round {} found when updating bucket {}
status to {}.",
+ rebalanceExecution.getCurrentRound(),
+ tableBucket,
+ rebalanceStatus);
+ return;
+ }
+ zkClient.registerRebalanceRound(
+ rebalanceRound
+ .get()
+ .withBucketStatus(tableBucket, planForBucket,
rebalanceStatus));
+ } catch (Exception e) {
+ throw new RebalanceFailureException(
+ String.format(
+ "Failed to update rebalance round status for
bucket %s. The root cause: %s",
+ tableBucket, e.getMessage()),
+ e);
+ }
Review Comment:
`updateCurrentRoundBucketStatus` throws `RebalanceFailureException` on
ZooKeeper write failures. Since this is called from `finishRebalanceTask`
(which is invoked from `CoordinatorEventProcessor` without a catch), a
transient ZooKeeper issue can bubble up and disrupt the coordinator event loop.
Prefer logging the failure (and optionally retrying later) instead of throwing
from this hot path.
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java:
##########
@@ -320,6 +383,8 @@ public void cancelRebalance(@Nullable String rebalanceId) {
LOG.error("Error when delete rebalance plan from zookeeper.", e);
}
Review Comment:
The error log message is both ungrammatical and misleading: this block is
not deleting anything; it's attempting to update the rebalance task znode.
Clarifying the message will make troubleshooting easier.
--
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]
