peter-toth commented on code in PR #59792:
URL: https://github.com/apache/spark/pull/59792#discussion_r4380697019
##########
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 =>
Review Comment:
**Finding 1.** The description says "Only `TaskCompletionListenerException`
is caught, so fatal errors still propagate." That holds only for an error that
`invokeListeners` throws itself. `invokeListeners` catches any `Throwable` from
a listener (`TaskContextImpl.scala:379`) and wraps it into the
`TaskCompletionListenerException`, so this catch swallows a listener's fatal
error too.
I checked it with a failure listener that throws `OutOfMemoryError`, next to
a failing interrupt listener:
- base: `markInterrupted` throws a `TaskCompletionListenerException` that
carries the OOM
- head: `markInterrupted` returns, and the OOM is only in the suppressed
exceptions of the task failure
The behaviour is fine, since it matches how `invokeListeners` treats every
listener failure. Could you reword the sentence, along the lines of the comment
in `markTaskCompleted` (`TaskContextImpl.scala:240-242`)? E.g. "Only
`TaskCompletionListenerException` is caught, so an error from `invokeListeners`
itself (e.g. an `OutOfMemoryError`) still propagates. A listener's own error is
wrapped into that exception, as for the interrupt listeners."
##########
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)
Review Comment:
**Finding 2.** This `logError` runs before the new `try`. #59794 hardens the
same case for pre-completion listeners: an appender that does not ignore
exceptions. Here that appender makes `markInterrupted` throw an
`AppenderLoggingException`. Then `Task.kill` skips `taskThread.interrupt()`, as
in the bug this PR fixes. `markTaskFailed` never runs either, so the task is
not marked failed and its failure listeners don't run. I checked it with an
appender that throws on this message. Base and head behave the same.
Making the log best effort fixes it. With this change the probe returns
normally and marks the task failed. The `TaskInterruptListener`, SPARK-56330
and SPARK-60554 tests in `TaskContextSuite` still pass:
```scala
try {
logError(log"Exception(s) in TaskInterruptListener(s) after task
was interrupted", e)
} catch {
case NonFatal(t) => e.addSuppressed(t)
}
```
(needs `import scala.util.control.NonFatal`)
The same appender failure on the failure listener's log line in
`invokeListeners` (`TaskContextImpl.scala:427`) escapes `markInterrupted` too.
It is not a `TaskCompletionListenerException`, so the new catch misses it. It
also leaves `listenerInvocationThread` set, so the completion listeners never
run. That part isn't specific to `markInterrupted`, so it may fit SPARK-60556
better.
--
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]
