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 0b1644a0436f [SPARK-60567][CORE] Keep running task interrupt listeners
if logging a listener failure throws
0b1644a0436f is described below
commit 0b1644a0436f490a9c2f50ee04e5ec15759cf33f
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 16:46:55 2026 -0700
[SPARK-60567][CORE] Keep running task interrupt listeners if logging a
listener failure throws
### What changes were proposed in this pull request?
This PR makes the two `logError` calls on the interrupt (kill) path of
`TaskContextImpl` best effort:
- In `invokeTaskInterruptListeners`, the per-listener `logError` in the
loop is wrapped in `try`/`catch` (`Throwable`, like the catch it sits in and
like the loops in #59794). A failure to log is added to the listener's
exception as a suppressed exception, and the loop goes on with the remaining
listeners.
- In `markInterrupted`, the `logError` of the aggregate failure is wrapped
in `try`/`catch` (`NonFatal`). A failure to log is added to the aggregate
exception that `markTaskFailed(e)` records right after. `NonFatal` keeps the
contract of SPARK-60554, where an error from `invokeListeners` itself (e.g. an
`OutOfMemoryError`) still propagates.
### Why are the changes needed?
With an appender that does not ignore exceptions
(`ignoreExceptions=false`), log4j throws an `AppenderLoggingException` from the
logging call. On the interrupt path this happens when an interrupt listener
fails and logging its failure fails:
- The loop ends, so the remaining interrupt listeners never run.
- The exception escapes `markInterrupted` before `markTaskFailed(e)`, so
the task is not marked failed (`isFailed` stays `false`) and its failure
listeners do not run.
- `Task.kill` calls `context.markInterrupted(reason)` before
`taskThread.interrupt()`, so the task thread is never interrupted.
SPARK-60554 (#59792) guarded only the `markTaskFailed(e)` call, and
SPARK-60556 (#59794) guarded the pre-completion loop and `invokeListeners`.
This PR covers the two remaining unguarded log calls on the interrupt path.
This change overlaps textually with any rewrite of
`invokeTaskInterruptListeners` that moves the recording of the aggregate
failure to the end of the loop. Whichever lands second should carry the
`try`/`catch` over to the moved `logError`.
### Does this PR introduce _any_ user-facing change?
No API or configuration change. Only deployments that use a log appender
that does not ignore exceptions are affected: a killed task whose interrupt
listener fails now still runs the remaining interrupt listeners, is marked
failed, and has its task thread interrupted. This behavior exists in released
Spark [4.2.0](https://github.com/apache/spark/releases/tag/v4.2.0) (2026-07-11).
### How was this patch tested?
Added three tests to `TaskContextSuite`. They reuse the `FailingAppender`
added by #59794, an appender with `ignoreExceptions=false` that throws for log
events whose message contains a trigger string:
- `SPARK-60567: all TaskInterruptListeners should be called even if logging
a failure fails`
- `SPARK-60567: markInterrupted marks the task failed even if logging the
listener failures fails`
- `SPARK-60567: Task.kill interrupts the task thread even if logging the
listener failures fails`
Without the fix, all three fail:
```
[info] - ... *** FAILED ***
[info] org.apache.logging.log4j.core.appender.AppenderLoggingException:
An exception occurred processing Appender failingAppender
```
The exception is thrown from `invokeTaskInterruptListeners`
(`TaskContextImpl.scala:323`), `markInterrupted` (`TaskContextImpl.scala:446`),
and `Task.kill`, respectively. With the fix, they pass, including 10
consecutive runs.
```
build/sbt 'core/testOnly *TaskContextSuite *BarrierTaskContextSuite
*ExecutorSuite *TaskSetManagerSuite'
build/sbt core/scalastyle core/Test/scalastyle
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5.5
Closes #59805 from dongjoon-hyun/SPARK-60567.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scala/org/apache/spark/TaskContextImpl.scala | 20 +++-
.../apache/spark/scheduler/TaskContextSuite.scala | 106 +++++++++++++++++++++
2 files changed, 124 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
index e619e78205ff..575d0e6f7512 100644
--- a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
+++ b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
@@ -23,6 +23,7 @@ import javax.annotation.concurrent.GuardedBy
import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
+import scala.util.control.NonFatal
import org.apache.spark.executor.TaskMetrics
import org.apache.spark.internal.{config, Logging}
@@ -327,7 +328,14 @@ private[spark] class TaskContextImpl(
} catch {
case e: Throwable =>
listenerExceptions += e
- logError(log"Error in TaskInterruptListener", e)
+ // Best effort: a failure to log, e.g. from an appender that does
not ignore exceptions,
+ // must not end the loop, or the remaining listeners would not run.
It is reported with
+ // `e` instead.
+ try {
+ logError(log"Error in TaskInterruptListener", e)
+ } catch {
+ case t: Throwable => e.addSuppressed(t)
+ }
}
}
if (listenerExceptions.nonEmpty) {
@@ -457,7 +465,15 @@ private[spark] class TaskContextImpl(
case e: TaskCompletionListenerException =>
// 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)
+ //
+ // Logging is best effort: a failure to log, e.g. from an appender
that does not ignore
+ // exceptions, must not skip `markTaskFailed(e)` below or escape. It
is reported with `e`
+ // instead.
+ try {
+ logError(log"Exception(s) in TaskInterruptListener(s) after task was
interrupted", e)
+ } catch {
+ case NonFatal(t) => e.addSuppressed(t)
+ }
try {
markTaskFailed(e)
} catch {
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 9132b71d1b9d..bc4e6cbe76ce 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
@@ -255,6 +255,112 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
}
}
+ test("SPARK-60567: all TaskInterruptListeners should be called even if
logging a failure " +
+ "fails") {
+ val context = TaskContext.empty()
+ val listener = mock(classOf[TaskInterruptListener])
+ val first = new Exception("exception in listener1")
+ context.addTaskInterruptListener(listener)
+ // Added last, so it runs first.
+ context.addTaskInterruptListener(new TaskInterruptListener {
+ override def onTaskInterrupted(context: TaskContext, reason: String):
Unit = throw first
+ })
+
+ val logFailure = new IllegalStateException("log failure")
+ withLogAppender(new FailingAppender("Error in TaskInterruptListener",
logFailure)) {
+ context.markInterrupted("test interrupt")
+ }
+ verify(listener, times(1)).onTaskInterrupted(any(), mockitoEq("test
interrupt"))
+ assert(context.isFailed())
+ val taskFailure = context.getTaskFailure.get
+ assert(taskFailure.isInstanceOf[TaskCompletionListenerException])
+ assert(taskFailure.getSuppressed.toSeq === Seq(first))
+ // The failure to log is reported with the failure it was logging.
+ assert(first.getSuppressed.length === 1)
+ val causes =
Iterator.iterate[Throwable](first.getSuppressed.head)(_.getCause)
+ assert(causes.takeWhile(_ != null).contains(logFailure))
+ }
+
+ test("SPARK-60567: markInterrupted marks the task failed even if logging the
listener " +
+ "failures fails") {
+ val context = TaskContext.empty()
+ val failureListenerCalls = new AtomicInteger(0)
+ val first = new Exception("exception in interrupt listener")
+ context.addTaskInterruptListener(new TaskInterruptListener {
+ override def onTaskInterrupted(context: TaskContext, reason: String):
Unit = throw first
+ })
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
=
+ failureListenerCalls.incrementAndGet()
+ })
+
+ val logFailure = new IllegalStateException("log failure")
+ withLogAppender(new FailingAppender("Exception(s) in
TaskInterruptListener(s)", logFailure)) {
+ // The failure to log must not be thrown from `markInterrupted` (kill /
RPC thread must not
+ // see it).
+ context.markInterrupted("test interrupt")
+ }
+ assert(context.isInterrupted())
+ assert(context.isFailed())
+ assert(failureListenerCalls.get() == 1)
+
+ // The task failure is the aggregate failure of the interrupt listeners
and reports the failure
+ // to log it.
+ val taskFailure = context.getTaskFailure.get
+ assert(taskFailure.isInstanceOf[TaskCompletionListenerException])
+ assert(taskFailure.getMessage.contains("exception in interrupt listener"))
+ assert(taskFailure.getSuppressed.contains(first))
+ assert(taskFailure.getSuppressed.exists { t =>
+ Iterator.iterate[Throwable](t)(_.getCause).takeWhile(_ !=
null).contains(logFailure)
+ })
+ }
+
+ test("SPARK-60567: Task.kill interrupts the task thread even if logging the
listener " +
+ "failures fails") {
+ 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 = {
+ // Both listeners fail, so both the per-listener and the aggregate
failure are logged.
+ (1 to 2).foreach { i =>
+ context.addTaskInterruptListener(new TaskInterruptListener {
+ override def onTaskInterrupted(context: TaskContext, reason:
String): Unit =
+ throw new Exception(s"exception in interrupt listener $i")
+ })
+ }
+ 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))
+ withLogAppender(
+ new FailingAppender("TaskInterruptListener", new
IllegalStateException("log failure"))) {
+ 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]
