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 22:13:53 UTC
(spark) branch branch-4.1 updated: [SPARK-60466][CORE] Fix `ThreadUtils.shutdown` to call `shutdownNow` after the grace period
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 3d9d1041822a [SPARK-60466][CORE] Fix `ThreadUtils.shutdown` to call `shutdownNow` after the grace period
3d9d1041822a is described below
commit 3d9d1041822ab6de7a4079390dbe05fb679f7252
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Oct 22 15:12:49 2026 -0700
[SPARK-60466][CORE] Fix `ThreadUtils.shutdown` to call `shutdownNow` after the grace period
### What changes were proposed in this pull request?
This PR aims to fix `ThreadUtils.shutdown` to call `shutdownNow()` when the executor does not terminate within the grace period.
Currently, `ThreadUtils.shutdown` checks `isShutdown` to decide whether to call `shutdownNow()`. However, `isShutdown` is always `true` after `shutdown()`, so `shutdownNow()` is never called.
```scala
executor.shutdown()
executor.awaitTermination(gracePeriod.toMillis, TimeUnit.MILLISECONDS)
if (!executor.isShutdown) { // Always false after `shutdown()`
executor.shutdownNow()
}
```
This PR uses the return value of `awaitTermination` instead, as in the usage example of [ExecutorService](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ExecutorService.html). In addition, if the calling thread is interrupted while waiting, this PR calls `shutdownNow()` before rethrowing the `InterruptedException`.
### Why are the changes needed?
The `shutdownNow()` branch has been dead code since `ThreadUtils.shutdown` was added by SPARK-24248 (apache/spark#21366). When tasks outlive the grace period, they keep running and the queued tasks are still executed. For a `ScheduledThreadPoolExecutor`, pending delayed tasks are kept until their delays elapse because it executes existing delayed tasks after `shutdown()` by default.
This became a regression in Apache Spark [v4.0.0](https://github.com/apache/spark/releases/tag/v4.0.0) (2025-05-19) because SPARK-46929 (apache/spark#44962) replaced the explicit `if (!awaitTermination(...)) shutdownNow()` logic of the following call sites with `ThreadUtils.shutdown`.
- `JobScheduler.stop`: A non-graceful `StreamingContext.stop` no longer interrupts running streaming jobs after 2 seconds.
- `CheckpointWriter.stop`: Pending checkpoint writes are no longer discarded after 10 seconds. Since its executor uses a non-daemon thread, the JVM cannot exit until all of them are written.
- `ElementTrackingStore.close`: Queued asynchronous cleanup tasks are no longer discarded after 5 seconds and can run against the closed store.
This also contributed to the JVM hang of SPARK-49479 (apache/spark#50020). The non-daemon `BarrierCoordinator` timer thread had a pending task with the 365-day `spark.barrier.sync.timeout`, and the thread was never terminated because `shutdownNow()` was not called after the 30-second grace period. `BarrierCoordinator` now cancels its timer tasks before shutting down, but `ThreadUtils.shutdown` itself still has the issue.
### Does this PR introduce _any_ user-facing change?
Yes. When an executor does not terminate within the grace period of `ThreadUtils.shutdown`, its running tasks are interrupted and its queued tasks are dropped. The same happens right away when the calling thread is interrupted while waiting. For the above three call sites, this restores the behavior of Apache Spark 3.5 in the grace period case. For the other callers, this is a new behavior because `shutdownNow()` was never called for them. All callers are in stop or close paths. Execu [...]
### How was this patch tested?
Pass the CIs with the newly added test cases.
Without this fix, three of the four new test cases fail.
- `SPARK-60466: shutdown interrupts running tasks and drops queued tasks after the grace period`
- `SPARK-60466: shutdown drops pending delayed tasks after the grace period`
- `SPARK-60466: shutdown forces termination when the calling thread is interrupted`
The remaining one, `SPARK-60466: shutdown waits for tasks that finish within the grace period`, passes with and without this fix to make sure that the graceful path is unchanged.
```
$ build/sbt "core/testOnly org.apache.spark.util.ThreadUtilsSuite"
...
[info] Tests: succeeded 15, 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 #59707 from dongjoon-hyun/SPARK-60466.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 7e9a966cf395de0e4c37451b4fca6be2c5aa7649)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../scala/org/apache/spark/util/ThreadUtils.scala | 19 +++-
.../org/apache/spark/util/ThreadUtilsSuite.scala | 105 +++++++++++++++++++++
2 files changed, 121 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala b/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
index d22e14d99265..8a0efeac7912 100644
--- a/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/ThreadUtils.scala
@@ -397,13 +397,26 @@ private[spark] object ThreadUtils {
}
// scalastyle:on awaitready
+ /**
+ * Shuts down the executor, waiting up to `gracePeriod` for running and queued tasks to finish.
+ * If they do not finish in time, or if the calling thread is interrupted while waiting, calls
+ * `shutdownNow()` to interrupt the running tasks and drop the queued ones. In the latter case,
+ * the `InterruptedException` is rethrown. This method does not wait again after
+ * `shutdownNow()`, so the executor may not be terminated yet when it returns, and tasks that
+ * ignore interrupts may keep running.
+ */
def shutdown(
executor: ExecutorService,
gracePeriod: Duration = FiniteDuration(30, TimeUnit.SECONDS)): Unit = {
executor.shutdown()
- executor.awaitTermination(gracePeriod.toMillis, TimeUnit.MILLISECONDS)
- if (!executor.isShutdown) {
- executor.shutdownNow()
+ try {
+ if (!executor.awaitTermination(gracePeriod.toMillis, TimeUnit.MILLISECONDS)) {
+ executor.shutdownNow()
+ }
+ } catch {
+ case e: InterruptedException =>
+ executor.shutdownNow()
+ throw e
}
}
diff --git a/core/src/test/scala/org/apache/spark/util/ThreadUtilsSuite.scala b/core/src/test/scala/org/apache/spark/util/ThreadUtilsSuite.scala
index d74bc2699944..b167bf506e6a 100644
--- a/core/src/test/scala/org/apache/spark/util/ThreadUtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/ThreadUtilsSuite.scala
@@ -132,6 +132,111 @@ class ThreadUtilsSuite extends SparkFunSuite {
}
}
+ test("SPARK-60466: shutdown interrupts running tasks and drops queued tasks after the " +
+ "grace period") {
+ val executor = ThreadUtils.newDaemonSingleThreadExecutor("ThreadUtilsSuite-shutdown")
+ try {
+ val started = new CountDownLatch(1)
+ val interrupted = new CountDownLatch(1)
+ @volatile var queuedTaskRan = false
+ executor.execute(() => {
+ started.countDown()
+ try {
+ new CountDownLatch(1).await()
+ } catch {
+ case _: InterruptedException => interrupted.countDown()
+ }
+ })
+ executor.execute(() => queuedTaskRan = true)
+ assert(started.await(10, TimeUnit.SECONDS))
+
+ ThreadUtils.shutdown(executor, 100.milliseconds)
+ assert(interrupted.await(10, TimeUnit.SECONDS))
+ eventually(timeout(10.seconds)) {
+ assert(executor.isTerminated)
+ }
+ assert(!queuedTaskRan)
+ } finally {
+ executor.shutdownNow()
+ }
+ }
+
+ test("SPARK-60466: shutdown drops pending delayed tasks after the grace period") {
+ val executor =
+ ThreadUtils.newDaemonSingleThreadScheduledExecutor("ThreadUtilsSuite-shutdown-delayed")
+ try {
+ // A scheduled executor still runs its delayed tasks after shutdown() by default, so only
+ // shutdownNow() can terminate it before the delay elapses.
+ executor.schedule(new Runnable {
+ override def run(): Unit = {}
+ }, 1, TimeUnit.HOURS)
+
+ ThreadUtils.shutdown(executor, 100.milliseconds)
+ eventually(timeout(10.seconds)) {
+ assert(executor.isTerminated)
+ }
+ } finally {
+ executor.shutdownNow()
+ }
+ }
+
+ test("SPARK-60466: shutdown waits for tasks that finish within the grace period") {
+ val executor = ThreadUtils.newDaemonSingleThreadExecutor("ThreadUtilsSuite-shutdown-graceful")
+ try {
+ @volatile var interrupted = false
+ @volatile var queuedTaskRan = false
+ executor.execute(() => {
+ try {
+ Thread.sleep(500)
+ } catch {
+ case _: InterruptedException => interrupted = true
+ }
+ })
+ executor.execute(() => queuedTaskRan = true)
+
+ ThreadUtils.shutdown(executor, 1.minute)
+ assert(executor.isTerminated)
+ assert(!interrupted)
+ assert(queuedTaskRan)
+ } finally {
+ executor.shutdownNow()
+ }
+ }
+
+ test("SPARK-60466: shutdown forces termination when the calling thread is interrupted") {
+ val executor =
+ ThreadUtils.newDaemonSingleThreadExecutor("ThreadUtilsSuite-shutdown-interrupted")
+ try {
+ val started = new CountDownLatch(1)
+ val interrupted = new CountDownLatch(1)
+ executor.execute(() => {
+ started.countDown()
+ try {
+ new CountDownLatch(1).await()
+ } catch {
+ case _: InterruptedException => interrupted.countDown()
+ }
+ })
+ assert(started.await(10, TimeUnit.SECONDS))
+
+ Thread.currentThread().interrupt()
+ try {
+ intercept[InterruptedException] {
+ ThreadUtils.shutdown(executor, 1.minute)
+ }
+ } finally {
+ // Do not leak the interrupt status to the following tests.
+ Thread.interrupted()
+ }
+ assert(interrupted.await(10, TimeUnit.SECONDS))
+ eventually(timeout(10.seconds)) {
+ assert(executor.isTerminated)
+ }
+ } finally {
+ executor.shutdownNow()
+ }
+ }
+
test("sameThread") {
val callerThreadName = Thread.currentThread().getName()
val f = Future {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org