You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "miaoever (via GitHub)" <gi...@apache.org> on 2026/10/04 00:40:49 UTC
[PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
miaoever opened a new pull request, #59229:
URL: https://github.com/apache/spark/pull/59229
### What changes were proposed in this pull request?
Update `ExecutorAllocationListener` so an explicitly killed, unfinished regular task is counted as pending again and restarts the scheduler backlog timer.
The listener records task indices completed by speculative attempts so that a later `TaskKilled` event for the corresponding regular attempt remains a no-op, preserving the behavior required by SPARK-30511. Killed speculative attempts and events from completed stages also remain no-ops.
### Why are the changes needed?
When a regular task starts while the executor target is zero, it can drain the scheduler queue and clear the backlog timer. If that task is killed before succeeding, the scheduler requeues it, but the listener currently ignores `TaskKilled`. The task is therefore not counted as pending and the timer is not restarted. If the available executor is then lost, dynamic allocation may not request a replacement and the stage can stall.
See [SPARK-59972](https://issues.apache.org/jira/browse/SPARK-59972).
### Does this PR introduce _any_ user-facing change?
Yes. Dynamic allocation now requests an executor for a requeued unfinished regular task after its previous attempt is explicitly killed.
### How was this patch tested?
Added `ExecutorAllocationManagerSuite` regression tests covering:
- an unfinished regular task being killed;
- a speculative attempt being killed;
- a regular attempt being killed after a speculative copy succeeds; and
- a killed-task event arriving after stage completion.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "miaoever (via GitHub)" <gi...@apache.org>.
miaoever commented on code in PR #59229:
URL: https://github.com/apache/spark/pull/59229#discussion_r4176343526
##########
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:
@HyukjinKwon Added a regression test covering a non-killed regular-task failure after its speculative copy succeeds. The test confirms the task index remains completed with zero pending demand. Also verified that the test fails when the successful-speculative guard is removed and passes with the guard restored. ([ExecutorAllocationManagerSuite.scala](https://github.com/tmp/agentcloud-100000578760927-workspace/6be51178-a1c4-4043-9980-828e45c11858/spark/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala:965), line 965)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "miaoever (via GitHub)" <gi...@apache.org>.
miaoever commented on PR #59229:
URL: https://github.com/apache/spark/pull/59229#issuecomment-6221715565
Hi @HyukjinKwon , May I know if you have more comments on this PR (I've addressed all existing comments above). Ty!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59229:
URL: https://github.com/apache/spark/pull/59229#discussion_r4227649704
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -950,12 +986,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 &&
+ !stageAttemptToSuccessfulSpeculativeTaskIndices.get(stageAttempt)
+ .exists(_.contains(taskIndex))) {
+ // An unsuccessful regular task remains pending until another attempt starts.
Review Comment:
**Nit (P3):** This comment doesn't match how the listener ends the pending state. The index goes back into `stageAttemptToTaskIndices` in two places:
- `onTaskStart`, but only for a regular attempt.
- the `Success` arm above (lines 953-954), for any successful attempt.
So a speculative copy that was already running ends the pending state when it succeeds, without any new attempt starting. The `stop counting a task as pending when speculation succeeds after $reason` tests rely on exactly that. A speculative attempt starting, on the other hand, does not end it. This accounting has been easy to get wrong across revisions, so the comment should name the real exits:
```suggestion
// The index stays pending until a regular attempt starts or any attempt succeeds.
```
**Verification:**
- **Inspection:** The revised comment names both exits implemented by the listener (a regular attempt start in onTaskStart and any successful attempt in the Success arm) and does not claim that a speculative attempt start ends the pending state.
- **Compile:** The core module still compiles and passes style checks, with the edited line within the 100-character limit and ASCII only.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59229:
URL: https://github.com/apache/spark/pull/59229#discussion_r4175787943
##########
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:
**Nit (P3):** This new guard -- skip the index removal when the speculative copy already succeeded -- is only reached on a non-killed failure, and no test drives that path. All four added tests post `TaskKilled` (the sibling arm), and the existing `SPARK-30511` test fails index 38 *before* its speculative copy succeeds, so it removes the index exactly as the old code did and never crosses this guard. A regression that dropped the guard would re-pend an already-completed index -- raising `maxNumExecutorsNeededPerResourceProfile` -- with every test still green. Worth adding a case where a regular task's speculative copy succeeds and the regular attempt then ends with a non-killed failure (e.g. `ExceptionFailure`/`UnknownReason`), asserting the index stays counted as done.
**Verification:**
- **Regression:** A test exercising the case _ => arm with a non-killed failure after a recorded speculative success fails when the successful-speculative guard is removed (index re-pended, maxNumExecutorsNeeded rises) and passes with the guard present.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59972][CORE] Count killed unfinished tasks as pending [spark]
Posted by "miaoever (via GitHub)" <gi...@apache.org>.
miaoever commented on code in PR #59229:
URL: https://github.com/apache/spark/pull/59229#discussion_r4176343526
##########
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:
@HyukjinKwon Added a regression test covering a non-killed regular-task failure after its speculative copy succeeds. The test confirms the task index remains completed with zero pending demand. I also verified that the test fails when the successful-speculative guard is removed and passes with the guard restored. ([ExecutorAllocationManagerSuite.scala](https://github.com/tmp/agentcloud-100000578760927-workspace/6be51178-a1c4-4043-9980-828e45c11858/spark/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala:965), line 965)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org