dongjoon-hyun commented on code in PR #60164:
URL: https://github.com/apache/spark/pull/60164#discussion_r4489843772
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -207,8 +207,21 @@ private[spark] class TaskContextImpl(
}
private[spark] def invokePostStatusUpdateListeners(): Unit = {
+ // The executor has already reported the task's success to the driver and
released the task's
+ // block locks and memory, and `Task.run` has unset its `TaskContext`. So
a failing listener
+ // must not mark the task failed, which `invokeListeners()` does for a
listener that throws:
+ // the failure listeners would run for a task that has not failed, without
those resources.
+ val listenerExceptions = new ArrayBuffer[Throwable](2)
invokeListeners(onPostStatusUpdateCallbacks, "PostStatusUpdateListener",
None, Claim) {
- _.onStatusUpdateSent(this)
+ listener =>
+ try {
+ listener.onStatusUpdateSent(this)
+ } catch {
+ case e: Throwable => listenerExceptions += e
Review Comment:
Good catch, thanks. Fixed in e934a6c: the sentence now says that the
listeners of `invokePostStatusUpdateListeners()` collect their failures
themselves, so they neither mark the task failed nor start the failure
listeners. It still holds for the other invocations, e.g. on a kill thread.
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -1988,6 +1988,41 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
assert(!completionInvoked) // completion listener should NOT be triggered
}
+ test("SPARK-60992: a failing post-status-update listener does not mark the
task failed") {
+ val context = TaskContext.empty()
+ val failureListener = mock(classOf[TaskFailureListener])
+ context.addTaskFailureListener(failureListener)
+ context.markTaskCompleted(None)
+
+ var isFailedInLastListener = true
+ context.addPostStatusUpdateListener(new PostStatusUpdateListener {
+ override def onStatusUpdateSent(context: TaskContext): Unit =
+ isFailedInLastListener = context.isFailed()
+ })
+ val error = new RuntimeException("post-status-update failure")
+ context.addPostStatusUpdateListener(new PostStatusUpdateListener {
+ override def onStatusUpdateSent(context: TaskContext): Unit = throw error
+ })
+
+ // The caller, the executor, still gets the failure to log it.
+ val e = intercept[TaskCompletionListenerException] {
+ context.invokePostStatusUpdateListeners()
+ }
+ assert(e.getMessage.contains("post-status-update failure"))
+ assert(e.getSuppressed.toSeq === Seq(error))
Review Comment:
Agreed, thanks. Fixed in e934a6c: the test now registers a second throwing
listener and asserts that the suppressed exceptions are both failures, in
invocation order (`Seq(anotherError, error)`).
--
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]
