LiebingYu commented on code in PR #3429:
URL: https://github.com/apache/fluss/pull/3429#discussion_r3355932518
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/rebalance/RebalanceManager.java:
##########
@@ -302,6 +367,8 @@ public RebalanceTask generateRebalanceTask(List<Goal>
goalsByPriority) {
private void processNewRebalanceTask() {
TableBucket tableBucket = inProgressRebalanceTasksQueue.peek();
if (tableBucket != null &&
inProgressRebalanceTasks.containsKey(tableBucket)) {
+ inflightTaskBucket = tableBucket;
+ inflightTaskStartMs = clock.milliseconds();
Review Comment:
Nit: inflightTaskBucket and inflightTaskStartMs rely on a fragile implicit
write-ordering invariant
Currently this is safe because every clear path writes inflightTaskStartMs =
-1 before clearing inflightTaskBucket, and processNewRebalanceTask() writes
inflightTaskBucket before inflightTaskStartMs = now(). Since checkTimeout()
reads inflightTaskStartMs first and returns immediately if it is -1, no
dangerous interleaving can occur today.
However, this correctness depends on the write order being maintained across
four separate call sites, with no compile-time enforcement. If a future change
swaps the write order in any of these paths (e.g., writes inflightTaskStartMs
before inflightTaskBucket in processNewRebalanceTask), the timeout thread could
read a stale bucket or a stale startMs and fire a spurious timeout for the
wrong task.
Consider combining both fields into a single immutable holder behind an
AtomicReference to make the invariant structurally unbreakable rather than
relying on write ordering:
```
private final AtomicReference<InflightTask> inflightTask = new
AtomicReference<>();
```
--
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]
