HyukjinKwon commented on code in PR #59229:
URL: https://github.com/apache/spark/pull/59229#discussion_r4225456372
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -938,10 +945,37 @@ private[spark] class ExecutorAllocationManager(
taskEnd.reason match {
case Success =>
+ if (taskEnd.taskInfo.speculative) {
Review Comment:
**Nit (P3):** The new `Resubmitted` arm removes the index from
`stageAttemptToTaskIndices` even when the regular attempt is still running, but
this branch adds it back only for a speculative `Success`. One ordering is left
uncovered. The speculative copy succeeds, so
`TaskSetManager.handleSuccessfulTask` kills the regular attempt. The
speculative executor is then lost before that kill lands, so `executorLost`
resets `successful(index)` and posts `Resubmitted`. If the regular attempt's
result is handled after that, `handleSuccessfulTask` accepts it as a real
`Success`, because the `successful(index) &&
killedByOtherAttempt.contains(tid)` check no longer matches. The listener never
adds the index back.
`getPendingTaskSum` then counts a completed task until the stage ends.
`hasPendingTasks` stays true, the add timer is never cleared, and
`maxNumExecutorsNeeded` is one task too high, which is the SPARK-30511
overcount. Before this PR, a speculative `Resubmitted` left the index in place,
so this ordering did not overcount. The window is narrow: the executor that
held the speculative output must be lost after the speculative success but
before the losing attempt's result is handled.
Adding the index back for every successful attempt closes it, and the add is
a no-op for a normal regular success. For example:
```scala
case Success =>
if (taskEnd.taskInfo.speculative) {
stageAttemptToSuccessfulSpeculativeTaskIndices.get(stageAttempt).foreach(_ +=
taskIndex)
}
// Any successful attempt completes the index, even after a failure, kill,
or resubmission.
val completedPending =
stageAttemptToTaskIndices.get(stageAttempt).exists(_.add(taskIndex))
stageAttemptToPendingSpeculativeTasks.get(stageAttempt).foreach(_.remove(taskIndex))
if (completedPending && !hasPendingTasks) {
allocationManager.onSchedulerQueueEmpty()
}
```
**Verification:**
- **Regression:** For the event sequence regular start, speculative start,
speculative Success, Resubmitted with the speculative TaskInfo, then regular
Success, the listener reports zero pending tasks, zero needed executors, and an
unset add timer.
- **Behavior:** The existing kill, failure, speculative-success,
resubmission, and completed-stage scenarios keep their expected pending counts,
running counts, and add-timer state.
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -938,10 +945,37 @@ private[spark] class ExecutorAllocationManager(
taskEnd.reason match {
case Success =>
+ if (taskEnd.taskInfo.speculative) {
+ stageAttemptToSuccessfulSpeculativeTaskIndices.get(stageAttempt)
+ .foreach(_ += taskIndex)
+ // The regular attempt may already have failed or been killed.
+ stageAttemptToTaskIndices.get(stageAttempt).foreach(_ +=
taskIndex)
+ }
// Remove pending speculative task in case the normal task
// is finished before starting the speculative task
stageAttemptToPendingSpeculativeTasks.get(stageAttempt).foreach(_.remove(taskIndex))
+ if (taskEnd.taskInfo.speculative && !hasPendingTasks) {
Review Comment:
**Nit (P3):** This new `onSchedulerQueueEmpty()` call is correct only
because of `!hasPendingTasks`: a speculative success must not clear the add
timer while other indexes or stages are still waiting for executors. No test
pins that half of the condition. Every new test that posts a speculative
`Success` and checks `addTime` uses a one-task stage (`createStageInfo(0, 1)`).
In the existing SPARK-30511 test, every speculative success happens after all
regular and speculative tasks have started, and that test never asserts
`addTime`.
If the guard were dropped, each speculative success would reset `addTime`
and the ramp-up counter while a backlog remains, and the suite would stay
green. That is the same kind of under-allocation this PR fixes for kills. A
case with a two-task stage, where a speculative copy succeeds while the other
index is still pending and `addTime` stays set, would cover it.
**Verification:**
- **Regression:** With another task index still pending, a speculative
Success leaves the add timer set and the pending count positive. Removing the
`!hasPendingTasks` condition makes the check fail.
--
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]
