You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/22 23:55:05 UTC
(spark) branch branch-4.x updated: [SPARK-60474][CORE] Do not wait for pending abort timer tasks in `TaskSchedulerImpl.stop`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 0ab28ba1164b [SPARK-60474][CORE] Do not wait for pending abort timer tasks in `TaskSchedulerImpl.stop`
0ab28ba1164b is described below
commit 0ab28ba1164b477af69df750fe52fa78b7a6e6be
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Oct 22 16:54:46 2026 -0700
[SPARK-60474][CORE] Do not wait for pending abort timer tasks in `TaskSchedulerImpl.stop`
### What changes were proposed in this pull request?
This PR aims to make `TaskSchedulerImpl.stop` discard the pending tasks of `abortTimer` without waiting, by calling `shutdownNow()` instead of `ThreadUtils.shutdown`.
```diff
ThreadUtils.shutdown(starvationTimer)
- ThreadUtils.shutdown(abortTimer)
+ // Discard pending abort tasks, delayed by `UNSCHEDULABLE_TASKSET_TIMEOUT`, without waiting.
+ abortTimer.shutdownNow()
```
When excludeOnFailure is enabled and a task set becomes unschedulable because all executors are excluded, `TaskSchedulerImpl` puts a one-shot task into `abortTimer` with the delay of `spark.scheduler.excludeOnFailure.unschedulableTaskSetTimeout` (default: `120s`). The task is not cancelled when the task set becomes schedulable again, so it stays in the queue until the delay elapses.
`DAGScheduler.stop` stops its `messageScheduler`, which also holds one-shot delayed tasks, with `shutdownNow()` in the same way. If an abort task happens to be running, `shutdownNow()` interrupts it, but the task only acquires the scheduler lock and posts an event to the `DAGScheduler`, and neither is interruptible.
### Why are the changes needed?
Since Apache Spark [v4.0.0](https://github.com/apache/spark/releases/tag/v4.0.0) (2025-05-19), `SparkContext.stop` can take up to 30 seconds longer because `TaskSchedulerImpl.stop` waits for such a pending abort task.
Previously, `abortTimer` was a `java.util.Timer`, and `stop` called `abortTimer.cancel()`, which discards all scheduled tasks without waiting. SPARK-46895 (apache/spark#44718) replaced it with a `ScheduledThreadPoolExecutor` and `ThreadUtils.shutdown(abortTimer)`. However, a `ScheduledThreadPoolExecutor` still executes the existing delayed tasks after `shutdown()` by default, so `ThreadUtils.shutdown` waits for its whole 30-second grace period.
The existing test cases are affected too. Before this PR, `afterEach` of `TaskSchedulerImplSuite` waited 70 seconds in total for the pending abort tasks of the following test cases.
- `SPARK-22148 try to acquire a new executor when task is unschedulable with 1 executor`: 10 seconds until its 10-second abort task runs.
- `SPARK-22148 abort timer should clear unschedulableTaskSetToExpiryTime for all TaskSets`: 60 seconds because `TaskSchedulerImpl.stop` is called twice, directly and via `DAGScheduler.stop`.
### Does this PR introduce _any_ user-facing change?
Yes, `SparkContext.stop` no longer waits up to 30 seconds for the pending abort timer tasks. This restores the behavior of Apache Spark 3.5. It only matters when excludeOnFailure is enabled and a task set became unschedulable within `spark.scheduler.excludeOnFailure.unschedulableTaskSetTimeout` before stopping.
### How was this patch tested?
Pass the CIs with the newly added test case.
Without this fix, the new test case fails because `stop` waits for 30 seconds.
```
[info] - SPARK-60474: stop should not wait for pending abort timer tasks *** FAILED *** (30 seconds, 60 milliseconds)
[info] 30005183833 was not less than 10000000000 (TaskSchedulerImplSuite.scala:1173)
```
With this fix, `TaskSchedulerImplSuite` passes in about 7 seconds. Without this fix, the existing test cases alone took about 77 seconds.
```
$ build/sbt "core/testOnly org.apache.spark.scheduler.TaskSchedulerImplSuite"
...
[info] - SPARK-60474: stop should not wait for pending abort timer tasks (27 milliseconds)
...
[info] Run completed in 6 seconds, 739 milliseconds.
[info] Total number of tests run: 122
[info] Tests: succeeded 122, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59712 from dongjoon-hyun/SPARK-60474.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 060a8d464058438ed89df934566aa94badacc480)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../apache/spark/scheduler/TaskSchedulerImpl.scala | 3 +-
.../spark/scheduler/TaskSchedulerImplSuite.scala | 34 ++++++++++++++++++++++
2 files changed, 36 insertions(+), 1 deletion(-)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
index d1eb646fc316..91c73b9a20ea 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
@@ -1035,7 +1035,8 @@ private[spark] class TaskSchedulerImpl(
}
}
ThreadUtils.shutdown(starvationTimer)
- ThreadUtils.shutdown(abortTimer)
+ // Discard pending abort tasks, delayed by `UNSCHEDULABLE_TASKSET_TIMEOUT`, without waiting.
+ abortTimer.shutdownNow()
}
override def defaultParallelism(): Int = backend.defaultParallelism()
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
index 810a0b7c483b..5b161db6c857 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala
@@ -1139,6 +1139,40 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext
}
}
+ test("SPARK-60474: stop should not wait for pending abort timer tasks") {
+ // Keep the abort timer task pending much longer than the 30s grace period of
+ // `ThreadUtils.shutdown`
+ taskScheduler = setupSchedulerWithMockTaskSetExcludelist(
+ config.UNSCHEDULABLE_TASKSET_TIMEOUT.key -> "1h")
+
+ // We have only 1 task remaining with 1 executor
+ val taskSet = FakeTask.createTaskSet(numTasks = 1)
+ taskScheduler.submitTasks(taskSet)
+ val tsm = stageToMockTaskSetManager(0)
+
+ // submit an offer with one executor
+ val firstTaskAttempts = taskScheduler.resourceOffers(IndexedSeq(
+ WorkerOffer("executor0", "host0", 1)
+ )).flatten
+
+ // Fail the running task
+ val failedTask = firstTaskAttempts.head
+ failTask(failedTask.taskId, TaskState.FAILED, UnknownReason, tsm)
+ when(tsm.taskSetExcludelistHelperOpt.get.isExecutorExcludedForTask(
+ "executor0", failedTask.index)).thenReturn(true)
+
+ // make an offer on the excluded executor. We won't schedule anything, and start the abort
+ // timer for the task set
+ assert(taskScheduler.resourceOffers(IndexedSeq(
+ WorkerOffer("executor0", "host0", 1)
+ )).flatten.isEmpty)
+ assert(taskScheduler.unschedulableTaskSetToExpiryTime.contains(tsm))
+
+ val startTime = System.nanoTime()
+ taskScheduler.stop()
+ assert(System.nanoTime() - startTime < 10.seconds.toNanos)
+ }
+
/**
* Helper for performance tests. Takes the explicitly excluded nodes and executors; verifies
* that the excluded are used efficiently to ensure scheduling is not O(numPendingTasks).
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org