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 2480ad033096 [SPARK-60556][CORE] Keep running task listeners if
logging a pre-completion listener failure throws
2480ad033096 is described below
commit 2480ad033096a86556c8f1cc958b2b604435e442
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Mon Oct 26 14:33:39 2026 -0700
[SPARK-60556][CORE] Keep running task listeners if logging a pre-completion
listener failure throws
### What changes were proposed in this pull request?
This PR makes the logging of a failed listener best effort in two places:
`TaskContextImpl.invokeTaskPreCompletionListener`, which runs the
pre-completion listeners, and `invokeListeners`, which runs the completion,
failure and post-status-update listeners. If `logError` throws, the failure is
added to the failure of the listener with `addSuppressed`, like the failure of
`markTaskFailed(e)` right above it, instead of ending the loop. It is then
reported by the `TaskCompletionListenerE [...]
Guarding a log call this way, where a later step must not be skipped, is
not new: see `SparkUncaughtExceptionHandler` before it halts the process,
`TaskMemoryManager.logPageAllocationFailure`, and the warnings of
`PipelinedWriterTask.awaitExit`.
`invokeTaskInterruptListeners` has the same unguarded `logError`, and so
does the one in `markInterrupted`. They are on the kill path, which this PR
does not touch, so they are not changed here.
apache/spark#59663 reworks `invokeListeners()` and keeps its `logError`
unguarded, so whichever lands second needs a rebase: the guard is the same five
lines around that `logError`. Both PRs also add a `FailingAppender` with the
same signature to `TaskContextSuite`.
### Why are the changes needed?
`markTaskCompleted()` runs the pre-completion listeners in a loop and sets
`completed` only when it finds none left. The completion listeners run after
that. `invokeTaskPreCompletionListener()` catches the failure of a listener,
but then logs it without a guard. If `logError` throws, the loop ends before
`completed` is set. This happens when an appender configured with
`ignoreExceptions=false` (the default is `true`) fails: log4j 2.26.1 then
throws an `AppenderLoggingException` to the [...]
The task is then stuck for good:
- the `markTaskCompleted(None)` that `runTaskWithListeners` calls in
`finally` returns at `if (completionStarted) return`,
- `addTaskCompletionListener()` only queues a listener while `completed` is
false, and
- `addTaskPreCompletionListener()` pushes onto a stack that nobody drains.
So the remaining pre-completion listeners, e.g. the one that stops a second
pipelined Python UDF writer (SPARK-60113, apache/spark#59332), and all
completion listeners never run. `isCompleted()` stays false, so
`BasePythonRunner.MonitorThread` keeps polling, and the executor frees the
memory of the task (`TaskMemoryManager.cleanUpAllAllocatedMemory()`) while a
pipelined writer may still use it, which is the use-after-free that SPARK-60113
prevents.
Setting `completed` in a `finally` would not do: the completion listeners
must not run before the remaining pre-completion listeners, and a
pre-completion listener left in the stack when `completed` is set would never
run.
`invokeListeners()` has the same flaw. A failing log line after a failing
listener ends its loop, so the remaining listeners do not run, and the calling
thread stays registered as the listener invocation thread, so no listener that
runs through `invokeListeners()` runs afterwards. This also stops the
completion listeners of a task whose pre-completion listener failed:
`markTaskFailed(e)` runs the failure listeners for it, and if one of them fails
and logging its failure fails, `invoke [...]
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests to `TaskContextSuite`, which use an appender that does not
ignore exceptions and fails the logging of a given message:
- A pre-completion listener fails, with a second pre-completion listener
and a completion listener behind it, and logging its failure fails. The
remaining listeners run in order, the task completes, and the failure to log is
reported with the failure of the listener.
- A completion listener fails, with another one behind it, and logging its
failure fails. The other one runs, a listener added afterwards runs right away,
and the failure to log is reported with the failure of the listener.
- A pre-completion listener fails, the failure listener that this runs
fails too, and logging that failure fails. The completion listener still runs
and the task completes.
Without the changes, the tests fail: the `AppenderLoggingException` escapes
`markTaskCompleted()`, or the completion listeners do not run. With the
changes, they pass.
```
build/sbt 'core/testOnly *TaskContextSuite'
build/sbt 'core/testOnly *BasePythonRunnerSuite'
build/sbt 'core/testOnly *ExecutorSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Sonnet 5.5
Closes #59794 from dongjoon-hyun/SPARK-60556.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scala/org/apache/spark/TaskContextImpl.scala | 18 +++-
.../apache/spark/scheduler/TaskContextSuite.scala | 98 ++++++++++++++++++++++
2 files changed, 114 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 56d32bc782d2..e619e78205ff 100644
--- a/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
+++ b/core/src/main/scala/org/apache/spark/TaskContextImpl.scala
@@ -268,7 +268,14 @@ private[spark] class TaskContextImpl(
case t: Throwable => e.addSuppressed(t)
}
listenerExceptions += e
- logError(log"Error in ${MDC(LISTENER, "TaskPreCompletionListener")}",
e)
+ // Best effort: a failure to log, e.g. from an appender that does not
ignore exceptions,
+ // must not end the loop in `markTaskCompleted()` before it sets
`completed`, or the
+ // completion listeners would never run. It is reported with `e`
instead.
+ try {
+ logError(log"Error in ${MDC(LISTENER,
"TaskPreCompletionListener")}", e)
+ } catch {
+ case t: Throwable => e.addSuppressed(t)
+ }
}
}
@@ -424,7 +431,14 @@ private[spark] class TaskContextImpl(
}
}
listenerExceptions += e
- logError(log"Error in ${MDC(LISTENER, name)}", 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
and this thread would
+ // stay registered as the listener invocation thread. It is reported
with `e` instead.
+ try {
+ logError(log"Error in ${MDC(LISTENER, name)}", e)
+ } catch {
+ case t: Throwable => e.addSuppressed(t)
+ }
}
}
if (listenerExceptions.nonEmpty) {
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 9c8edcefcdd6..9132b71d1b9d 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala
@@ -24,6 +24,9 @@ import java.util.concurrent.atomic.{AtomicInteger,
AtomicReference}
import scala.collection.mutable.ArrayBuffer
import scala.jdk.CollectionConverters._
+import org.apache.logging.log4j.core.LogEvent
+import org.apache.logging.log4j.core.appender.AbstractAppender
+import org.apache.logging.log4j.core.config.Property
import org.mockito.ArgumentMatchers.{any, eq => mockitoEq}
import org.mockito.Mockito._
import org.scalatest.BeforeAndAfter
@@ -966,6 +969,88 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
}
}
+ test("SPARK-60556: listeners still run if logging a pre-completion
listener's failure fails") {
+ val context = TaskContext.empty()
+ val invocationOrder = ArrayBuffer.empty[String]
+
+ def makeLoggingListener(id: String): TaskCompletionListener = new
TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = {
+ invocationOrder += id
+ }
+ }
+ val pre = new Exception("pre")
+ context.addTaskCompletionListener(makeLoggingListener("A"))
+ context.addTaskPreCompletionListener(makeLoggingListener("P2"))
+ // Added last, so it runs first.
+ context.addTaskPreCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw pre
+ })
+
+ val logFailure = new IllegalStateException("log failure")
+ withLogAppender(new FailingAppender("Error in TaskPreCompletionListener",
logFailure)) {
+ val e = intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ assert(e.getSuppressed.toSeq === Seq(pre))
+ // The failure to log is reported with the failure it was logging.
+ assert(pre.getSuppressed.length === 1)
+ val causes =
Iterator.iterate[Throwable](pre.getSuppressed.head)(_.getCause)
+ assert(causes.takeWhile(_ != null).contains(logFailure))
+ }
+ // The pre-completion listener after the failing one still runs, and
before the completion one.
+ assert(invocationOrder === Seq("P2", "A"))
+ assert(context.isCompleted())
+ }
+
+ test("SPARK-60556: completion listeners still run if logging a listener's
failure fails") {
+ val context = TaskContext.empty()
+ val invocations = new ConcurrentLinkedQueue[(String, Thread)]()
+ val first = new Exception("first")
+ context.addTaskCompletionListener(loggingListener(invocations, "B"))
+ // Added last, so it runs first.
+ context.addTaskCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw first
+ })
+
+ val logFailure = new IllegalStateException("log failure")
+ withLogAppender(new FailingAppender("Error in TaskCompletionListener",
logFailure)) {
+ val e = intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ assert(e.getSuppressed.toSeq === Seq(first))
+ // The failure to log is reported with the failure it was logging.
+ assert(first.getSuppressed.length === 1)
+ }
+ assert(invocations.asScala.map(_._1).toSeq === Seq("B"))
+ // This thread no longer counts as the one invoking listeners: a listener
added now runs
+ // right away.
+ context.addTaskCompletionListener(loggingListener(invocations, "C"))
+ assert(invocations.asScala.map(_._1).toSeq === Seq("B", "C"))
+ }
+
+ test("SPARK-60556: completion listeners run if logging a failure listener's
failure fails") {
+ val context = TaskContext.empty()
+ val invocations = new ConcurrentLinkedQueue[(String, Thread)]()
+ context.addTaskCompletionListener(loggingListener(invocations, "A"))
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
=
+ throw new Exception("failure listener")
+ })
+ // Its failure runs the failure listener, which fails too.
+ context.addTaskPreCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw new
Exception("pre")
+ })
+
+ val logFailure = new IllegalStateException("log failure")
+ withLogAppender(new FailingAppender("Error in TaskFailureListener",
logFailure)) {
+ intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ }
+ assert(invocations.asScala.map(_._1).toSeq === Seq("A"))
+ assert(context.isCompleted())
+ }
+
/** A completion listener that records its id and the thread it runs on in
`invocations`. */
private def loggingListener(
invocations: ConcurrentLinkedQueue[(String, Thread)],
@@ -1004,6 +1089,19 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
completion
}
+ /**
+ * An appender that throws `failure` for every event whose message contains
`trigger`. Unlike
+ * `LogAppender`, it does not ignore exceptions, so log4j throws them from
the logging call.
+ */
+ private class FailingAppender(trigger: String, failure: RuntimeException)
+ extends AbstractAppender("failingAppender", null, null, false,
Property.EMPTY_ARRAY) {
+ override def append(event: LogEvent): Unit = {
+ if (event.getMessage.getFormattedMessage.contains(trigger)) {
+ throw failure
+ }
+ }
+ }
+
test("SPARK-46480: Add isFailed in TaskContext") {
val context = TaskContext.empty()
var isFailed = false
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
