HyukjinKwon commented on code in PR #59229:
URL: https://github.com/apache/spark/pull/59229#discussion_r4218770111
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -950,12 +968,10 @@ private[spark] class ExecutorAllocationManager(
// (SPARK-8366)
allocationManager.onSchedulerBacklogged()
}
- if (!taskEnd.taskInfo.speculative) {
- // If a non-speculative task is intentionally killed, it means
the speculative task
- // has succeeded, and no further task of this task index will be
resubmitted. In this
- // case, the task index is completed and we shouldn't remove it
from
- // stageAttemptToTaskIndices. Otherwise, we will have a pending
non-speculative task
- // for the task index (SPARK-30511)
+ if (!taskEnd.taskInfo.speculative &&
Review Comment:
**Non-blocking (P2):** `stageAttemptToSuccessfulSpeculativeTaskIndices` is
cleared only at stage completion, but a speculative success can be revoked
while the stage attempt is still running. In a shuffle map stage without an
external shuffle service (for example with shuffle tracking), or on
decommission, losing the executor that ran the successful speculative copy
triggers `TaskSetManager.executorLost`. It resets `successful(index)`,
re-queues the index, and posts `Resubmitted` with the speculative `TaskInfo`,
and the `case _` arm leaves the index alone because that info is speculative.
A new regular attempt of the index then starts. If it fails (for example
with `ExecutorLostFailure`), `TaskSetManager` re-queues it again, but this
guard still finds the stale entry and keeps the index in
`stageAttemptToTaskIndices`, so it is not counted as pending. Before this PR
the index was removed here. If this is the stage's remaining work and no
executor is left, `maxNumExecutorsNeeded` drops to zero and the target falls to
`minExecutors`. That is the same stall this PR fixes for kills.
Clearing the entry when the index is resubmitted would close it, for example
at the top of the `case _` arm:
```scala
if (taskEnd.reason == Resubmitted) {
stageAttemptToSuccessfulSpeculativeTaskIndices.get(stageAttempt)
.foreach(_.remove(taskIndex))
}
```
`Resubmitted` is a better hook than a later regular `TaskStart`. It also
covers a regular attempt whose kill is acknowledged only after the
resubmission, which `TaskSetManager` re-queues too.
**Verification:**
- **Regression:** For the event sequence speculative Success, TaskKilled for
the regular attempt, Resubmitted with the speculative TaskInfo, a new regular
TaskStart, and then a non-killed failure of that attempt, the listener counts
the index as pending and needs a non-zero number of executors.
- **Behavior:** A non-killed failure or kill of a regular attempt after a
speculative success with no resubmission still leaves zero pending tasks.
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -938,10 +943,23 @@ private[spark] class ExecutorAllocationManager(
taskEnd.reason match {
case Success =>
+ if (taskEnd.taskInfo.speculative) {
Review Comment:
**Nit (P3):** Suppose a regular attempt is killed (for example through
`SparkContext.killTaskAttempt`) while its speculative copy is still running.
The new `TaskKilled` arm removes the index from `stageAttemptToTaskIndices`, so
it counts as pending. `TaskSetManager` re-queues the index, but
`dequeueTaskFromList` won't launch a regular copy while `copiesRunning(index)
== 1`. If the speculative copy then succeeds, `successful(index)` is set and
the index never runs again.
This branch only records the index in
`stageAttemptToSuccessfulSpeculativeTaskIndices` and never puts it back into
`stageAttemptToTaskIndices`. `getPendingTaskSum` therefore keeps counting a
task that will never run until the stage completes. `hasPendingTasks` stays
true and the executor target stays one task too high, which is the
idle-executor holding SPARK-30511 fixed. Before this PR the kill was ignored,
so this ordering was fine. The existing failure arm already has the same gap
for non-killed failures.
Re-adding the index once its speculative copy succeeds would close it:
```scala
if (taskEnd.taskInfo.speculative) {
stageAttemptToSuccessfulSpeculativeTaskIndices.get(stageAttempt)
.foreach(_ += taskIndex)
// The index is complete; stop counting it as pending even if its regular
// attempt was already killed or failed while this copy was running.
stageAttemptToTaskIndices.get(stageAttempt).foreach(_ += taskIndex)
}
```
**Verification:**
- **Regression:** For the event sequence regular start, speculative start,
TaskKilled for the regular attempt, then speculative Success, the listener
reports zero pending tasks and zero needed executors.
- **Behavior:** The existing kill, speculative-kill, post-success kill,
post-success failure, and completed-stage scenarios keep their expected pending
counts and add-timer state.
--
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]
