FrankChen021 commented on code in PR #20299:
URL: https://github.com/apache/druid/pull/20299#discussion_r4017174595
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java:
##########
@@ -1123,6 +1144,17 @@ private void pendingTasksExecutionLoop()
break;
}
+ if (eligibleWorkers == null) {
+ eligibleWorkers =
ImmutableMap.copyOf(getWorkersEligibleToRunTasks());
+ }
+
+ // Cheap pre-filter against the reused snapshot: skip tasks that
clearly have no worker.
+ if (findWorkerToRunTask(ti.getTask(), eligibleWorkers) == null) {
+ continue;
+ }
+
+ // A candidate exists in the snapshot; re-select against a fresh
snapshot before
+ // reserving so the assignment reflects current worker capacity
(see comment above).
immutableWorker = findWorkerToRunTask(ti.getTask());
Review Comment:
[P2] Revalidate capacity atomically with reservation
**Finding:** This fresh selection is still not atomic with the reservation
below it. WorkerHolder can publish a new task announcement into
tasksSnapshotRef without statusLock, so a worker can become full after this
call returns but before workersWithUnacknowledgedTask.putIfAbsent runs. The
selected ImmutableWorkerInfo is then stale, and runTaskOnWorker only checks
removal/lazy/blacklist before posting /assignTask, allowing an oversubscribing
assignment for a worker whose capacity changed in that window.
**Suggestion:** Make capacity publication, the capacity check, and the
reservation atomic, or revalidate the selected worker immediately before or
inside assignment, and add a regression test for an announcement arriving
between selection and reservation.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunner.java:
##########
@@ -1123,6 +1144,17 @@ private void pendingTasksExecutionLoop()
break;
}
+ if (eligibleWorkers == null) {
+ eligibleWorkers =
ImmutableMap.copyOf(getWorkersEligibleToRunTasks());
+ }
+
+ // Cheap pre-filter against the reused snapshot: skip tasks that
clearly have no worker.
+ if (findWorkerToRunTask(ti.getTask(), eligibleWorkers) == null) {
Review Comment:
[P2] Invalidate the cached snapshot after a failed refresh
**Finding:** If the cached pre-filter snapshot contains a candidate but the
fresh selection at line 1158 returns null because that worker became
unavailable, eligibleWorkers remains unchanged and the loop continues. Every
remaining pending task then passes the stale pre-filter and calls
findWorkerToRunTask(task) again, rebuilding the full worker/task-announcement
snapshot while holding statusLock. A capacity or state change during a large
backlog can therefore reproduce the original O(pendingTasks x workers x
tasksAnnouncedPerWorker) lock stall.
**Suggestion:** Refresh or replace the cached pre-filter snapshot after a
fresh selection returns no worker, or stop the pass and wait for the
worker-state notification instead of repeating the expensive refresh for every
pending task.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
