dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4339138517
##########
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
+ }
+ }
+
+ // Chooses the next thing to do under the TaskContext lock, to be done
outside of it.
+ def nextStep(): Option[() => Unit] = synchronized {
+ if (!invokesFailureListeners && failureCauseOpt.isDefined &&
!onFailureCallbacks.empty()) {
+ val cause = failureCauseOpt.get
+ Some(() => runLeftListeners(() => invokeTaskFailureListeners(cause,
Nested)))
+ } else if (!listeners.empty()) {
+ val listener = listeners.pop()
+ Some(() => runListener(listener))
+ } else if (mode != Nested && nextListenerInvocationThread.isEmpty &&
completed &&
Review Comment:
**5. [Low] The post-status-update and kill-thread invocations now run
leftover listeners without the task's `TaskContext`, and no test covers it**
`invokePostStatusUpdateListeners` (L204-208) claims like any other
invocation, so it now also runs pending failure listeners (L589-591) and
leftover completion listeners (here). It runs on the task thread after
`Task.run` has called `TaskContext.unset()` (`Task.scala:166`) and after the
executor has sent `FINISHED` (`Executor.scala:1051`), and a kill thread that
holds the invocation has no `TaskContext` either. Some completion listeners
read `TaskContext.get()` instead of their argument, e.g.
`TungstenAggregationIterator.scala:497` and
`ObjectAggregationIterator.scala:91`, and a helper thread with the task's
`TaskContext`, such as `PipedRDD`'s stdin writer, can register them lazily
after completion. Left to one of these invocations, they now fail with an NPE
where master dropped them. On the kill path that failure then escapes
`markInterrupted` (L641) to the kill caller, as a failing failure listener
already could.
The SPARK-57491 test only checks a context that never completes, so nothing
covers these branches outside `markTaskCompleted`. Also, the new scaladoc at
L483-484 ("they queue them and leave them to this thread") does not hold for
post-status-update listeners, which are still dropped when another thread holds
the invocation, as the description says.
Could a test cover the post-status-update invocation of a completed context,
and could the scaladoc mention both points?
--
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]
