dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4339138480
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -412,21 +566,61 @@ private[spark] class TaskContextImpl(
// [[markTaskFailed]] here will have no effect and we simply
resume running the
// remaining completion handlers.
try {
- listenerInvocationThread = None
- markTaskFailed(e)
+ markTaskFailed(e, Nested)
} catch {
case t: Throwable => e.addSuppressed(t)
- } finally {
- synchronized {
- if (listenerInvocationThread.isEmpty) {
- listenerInvocationThread = Some(Thread.currentThread())
- }
- }
}
listenerExceptions += e
logError(log"Error in ${MDC(LISTENER, name)}", e)
}
}
+
+ def runLeftListeners(invoke: () => Unit): Unit = {
+ try {
+ invoke()
+ } catch {
+ // The nested invocation has logged each listener error.
+ case e: Throwable => listenerExceptions += e
Review Comment:
**2. [Low] `runLeftListeners` nests the inner
`TaskCompletionListenerException` and hides fatal errors**
This adds the nested invocation's whole `TaskCompletionListenerException` as
one listener error, which is the shape that my item 20 on #59332 removed from
`markTaskCompleted`. It takes only one thread: when the task body fails with
`e`, a completion listener that calls `addTaskFailureListener(f)` finds the
invocation held by this thread (L514-516), so L591 runs `f` here. If `f`
throws, the nested invocation throws a `TaskCompletionListenerException` with
`f`'s failure and `Some(e)`, and L625-627 wraps that in another one with
`Some(e)` again. Its message then has the `Previous exception in task: ...`
block with `e`'s stack trace twice, `getSuppressed` holds the wrapper instead
of `f`'s exception, and if a pre-completion listener also failed, L301 merges
the wrapper as a single entry.
The catch-all also takes errors that are not listener failures.
`runListener` already catches whatever a listener throws, so anything else here
comes from the nested invocation itself, e.g. an `OutOfMemoryError` in
`logError` (L574) or while it builds its exception. That error then ends up in
`getSuppressed`, which `Executor.isFatalError` does not follow, so the executor
keeps running. My item 28 avoided exactly that at L297-300.
Could this merge the listeners' own failures like L301, and let other errors
propagate through the release in the `finally` (L615-622)?
```scala
case e: TaskCompletionListenerException => listenerExceptions ++=
e.getSuppressed.toSeq
```
The nested failure invocation's `error` would then drop out of the message,
but it is the outer `error` when the task body failed, and otherwise still in
`getTaskFailure`. "SPARK-60222: completion listeners left to a failure invoker
run before it releases" passes either way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
