This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 6000a5639f63 [SPARK-60554][CORE] Do not propagate failure listener
exceptions from `TaskContextImpl.markInterrupted`
6000a5639f63 is described below
commit 6000a5639f63d45ed1d900f2e0ec20654d91f438
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 11:36:40 2026 -0700
[SPARK-60554][CORE] Do not propagate failure listener exceptions from
`TaskContextImpl.markInterrupted`
### What changes were proposed in this pull request?
This PR guards the `markTaskFailed(e)` call in
`TaskContextImpl.markInterrupted`. If a task failure listener throws while the
failure of the interrupt listeners is being recorded, the resulting
`TaskCompletionListenerException` is no longer propagated. It is attached to
the recorded task failure as a suppressed exception instead. Only
`TaskCompletionListenerException` is caught, so an error from `invokeListeners`
itself (e.g. an `OutOfMemoryError`) still propagates. A listener's own e [...]
### Why are the changes needed?
SPARK-56330 made `markInterrupted` swallow the exceptions of failing
`TaskInterruptListener`s and record them with `markTaskFailed(e)`.
- https://github.com/apache/spark/pull/55151
That call runs the task failure listeners, and `invokeListeners` throws a
new `TaskCompletionListenerException` when one of them fails. The exception
escapes `markInterrupted`:
- `Task.kill` calls `context.markInterrupted(reason)` before
`taskThread.interrupt()`, so the interrupt is skipped and a task thread blocked
in an interruptible wait or I/O is never woken.
- With `spark.task.reaper.enabled=true`, the kill runs inside
`TaskReaper.run`, where the exception reaches the uncaught exception handler
and the executor JVM exits. Without the reaper, the RPC `Inbox` only logs
"Ignoring error".
- `createResourceUninterruptibly` calls `markInterrupted` the same way and
skips its own `Thread.interrupt()`.
- It contradicts the documentation of
`TaskContext.addTaskInterruptListener`, which says exceptions are not
propagated from `markInterrupted`.
The failure is already recorded by `markTaskFailed` before the listeners
run, and each failing listener is already logged by `invokeListeners`, so
nothing is lost by not propagating it.
### Does this PR introduce _any_ user-facing change?
No API or configuration change. In the rare case where an interrupt
listener and a task failure listener both throw, killing a task now interrupts
the task thread as documented, instead of failing with
`TaskCompletionListenerException` (and, with the task reaper enabled, exiting
the executor JVM). This behavior exists in released Spark 4.2.0.
### How was this patch tested?
Added two tests to `TaskContextSuite`. Both fail without the fix with
`TaskCompletionListenerException: exception in failure listener` thrown from
`markInterrupted` (and from `Task.kill`), and pass with it:
- `SPARK-60554: a failing TaskFailureListener does not escape
markInterrupted`
- `SPARK-60554: Task.kill interrupts the task thread even if interrupt and
failure listeners fail`
```
build/sbt 'core/testOnly *TaskContextSuite *BarrierTaskContextSuite
*ExecutorSuite *TaskSetManagerSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5.5
Closes #59792 from dongjoon-hyun/SPARK-60554.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scala/org/apache/spark/TaskContextImpl.scala | 11 +++-
.../apache/spark/scheduler/TaskContextSuite.scala | 73 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 1 deletion(-)
diff --git a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
index dafc938a65c1..56d32bc782d2 100644
--- a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
+++ b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
@@ -444,7 +444,16 @@ private[spark] class TaskContextImpl(
// SPARK-56330: Do not propagate to the executor kill / RPC thread;
record the aggregate
// failure for the task (not the first listener exception only).
logError(log"Exception(s) in TaskInterruptListener(s) after task was
interrupted", e)
- markTaskFailed(e)
+ try {
+ markTaskFailed(e)
+ } catch {
+ case t: TaskCompletionListenerException =>
+ // A task failure listener failed too. The task is already marked
failed and each
+ // failing listener is logged by `invokeListeners()`. Do not
propagate this either, or
+ // the caller skips interrupting the task thread (and, with the
task reaper, the
+ // executor JVM exits on the uncaught exception).
+ e.addSuppressed(t)
+ }
}
}
diff --git
a/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
b/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
index a0b59f453a6e..9c8edcefcdd6 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
@@ -179,6 +179,79 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
assert(e.getMessage.contains("test interrupt"))
}
+ test("SPARK-60554: a failing TaskFailureListener does not escape
markInterrupted") {
+ val context = TaskContext.empty()
+ val failureListenerCalls = new AtomicInteger(0)
+ context.addTaskInterruptListener(new TaskInterruptListener {
+ override def onTaskInterrupted(context: TaskContext, reason: String):
Unit =
+ throw new Exception("exception in interrupt listener")
+ })
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= {
+ failureListenerCalls.incrementAndGet()
+ throw new Exception("exception in failure listener")
+ }
+ })
+
+ // Neither the interrupt listener failure nor the failure listener failure
it triggers may be
+ // thrown from `markInterrupted` (kill / RPC thread must not see them).
+ context.markInterrupted("test interrupt")
+ assert(context.isInterrupted())
+ assert(context.isFailed())
+ assert(failureListenerCalls.get() == 1)
+
+ // The task failure is the interrupt listener failure and keeps the
failure listener failure.
+ val taskFailure = context.getTaskFailure.get
+ assert(taskFailure.isInstanceOf[TaskCompletionListenerException])
+ assert(taskFailure.getMessage.contains("exception in interrupt listener"))
+ assert(taskFailure.getSuppressed.exists(
+ _.getMessage.contains("exception in failure listener")))
+ }
+
+ test("SPARK-60554: Task.kill interrupts the task thread even if interrupt
and " +
+ "failure listeners fail") {
+ sc = new SparkContext("local", "test")
+ val taskStarted = new CountDownLatch(1)
+ val threadInterrupted = new CountDownLatch(1)
+ val serializedTaskMetrics =
+
SparkEnv.get.closureSerializer.newInstance().serialize(TaskMetrics.registered).array()
+ val task = new Task[Int](0, 0, 0, 1, JobArtifactSet.emptyJobArtifactSet,
+ serializedTaskMetrics = serializedTaskMetrics) {
+ override def runTask(context: TaskContext): Int = {
+ context.addTaskInterruptListener(new TaskInterruptListener {
+ override def onTaskInterrupted(context: TaskContext, reason:
String): Unit =
+ throw new Exception("exception in interrupt listener")
+ })
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable):
Unit =
+ throw new Exception("exception in failure listener")
+ })
+ taskStarted.countDown()
+ try {
+ // Block until `Task.kill` interrupts this thread.
+ new CountDownLatch(1).await(10, TimeUnit.SECONDS)
+ } catch {
+ case _: InterruptedException => threadInterrupted.countDown()
+ }
+ 0
+ }
+ }
+ val taskThread = new Thread(
+ () => Utils.tryLogNonFatalError { task.run(0, 0, null, 1, null,
Option.empty) },
+ "task-kill-test")
+ taskThread.start()
+ try {
+ assert(taskStarted.await(10, TimeUnit.SECONDS))
+ task.kill(interruptThread = true, "test kill")
+ assert(threadInterrupted.await(10, TimeUnit.SECONDS))
+ assert(task.context.isInterrupted())
+ assert(task.context.isFailed())
+ } finally {
+ taskThread.interrupt()
+ taskThread.join(10000)
+ }
+ }
+
test("immediately call an interrupt listener if the context is already
interrupted") {
var invocations = 0
var lastReason: String = null
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
