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:
`processNewRebalanceTask()` writes `inflightTaskBucket` first, then
`inflightTaskStartMs`. Meanwhile, `checkTimeout()` (running on the scheduled
executor thread) reads `inflightTaskStartMs` first, then `inflightTaskBucket`.
I think in this case maybe some problem:
1. Coordinator thread finishes task tb1 (clears inflight state), then calls
processNewRebalanceTask() for tb2, writes inflightTaskBucket = tb2 but hasn't
written inflightTaskStartMs yet.
2. Timeout thread reads the stale inflightTaskStartMs (from tb1, already >
2min ago), then reads the new inflightTaskBucket = tb2.
3. Result: tb2, which just started, gets incorrectly timed out.
--
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]
