dongjoon-hyun commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4424194402
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -449,6 +654,52 @@ private[spark] class TaskContextImpl(
}
}
}
+
+ def runLeftFailureListeners(cause: Throwable): Unit = {
+ try {
+ invokeTaskFailureListeners(cause, Nested)
+ } catch {
+ // Merge the listeners' own failures, which the nested invocation has
logged, not the
+ // exception that wraps them. A fatal error (e.g. an OutOfMemoryError)
propagates, through
+ // the release below.
+ case e: TaskCompletionListenerException => listenerExceptions ++=
e.getSuppressed.toSeq
+ case NonFatal(t) => listenerExceptions += t
Review Comment:
**3. [Nit] A failing `getMessage` drops the other listeners' failures from
the report**
Round 3 handles a failing `getMessage` where the error is caught (L322,
L364, L644 and this line), but the error comes from one place.
`listenerExceptions.map(_.getMessage)` (L381, L534, L705) throws at the first
such exception, and then none of the collected failures is reported.
For example, two failure listeners are left to the completing thread. One
throws `new Exception("a")`, and the other an exception whose `getMessage`
throws. L705 throws, and this line adds only that error, so "a" is in the log
but not in what `markTaskCompleted` throws. "an exception whose getMessage
throws does not end the listener loops" pins this result for a single listener
(`TaskContextSuite.scala:1500`). Likewise, one pre-completion listener with
such an exception replaces all the failures that L339 was about to report.
Could the helper at L378-384 keep them, with L533-536 and L704-707 calling
it too?
```scala
private def newTaskCompletionListenerException(
listenerExceptions: Seq[Throwable],
error: Option[Throwable]): TaskCompletionListenerException = {
// `getMessage` is listener code. Report its failure in place of the
exception it fails for:
// `printStackTrace`, e.g. in `Utils.exceptionString`, would fail on
that exception as well.
val reported = listenerExceptions.map { e =>
try { e.getMessage; e } catch { case t if !Utils.isFatalError(t) => t }
}
val exception = new
TaskCompletionListenerException(reported.map(_.getMessage), error)
reported.foreach(exception.addSuppressed)
exception
}
```
A failing `getMessage` then no longer leaves an invocation, which also
settles item 1 for that path.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -218,30 +244,96 @@ private[spark] class TaskContextImpl(
// writer that adds listeners lazily while it pulls its input. And since
the completion
// listeners only run after `completed` is set, they run after every
pre-completion listener,
// even if another thread is invoking listeners at this point.
+ //
+ // The completion listeners then run on this thread too, and this method
returns only after
+ // they have run. If it returned while another thread is invoking
listeners (e.g. a kill thread
+ // running failure listeners), that thread would not run them, and the
task would go on to
+ // release its memory with them pending. So the listener invocation is
claimed together with
+ // setting `completed`, which also keeps a thread that adds a completion
listener at that
+ // moment from running them; if another thread holds it, that thread hands
it over to this one
+ // when done. Failure listeners that another thread leaves to this
invocation run before the
+ // next completion listener (see `invokeListeners()`). A completion
listener added after this
+ // method has run the others runs on the thread adding it, unless another
thread is invoking
+ // listeners at that moment: that thread runs it only if it is running
completion listeners
+ // (see `invokeListeners()`).
+ //
+ // The wait for the other thread has no time bound: if its listeners
block, so does this task.
+ // It logs a warning periodically, the first one with that thread's stack
trace. The task waits
+ // here before it releases its block locks and memory (the executor does
that once `Task.run`
+ // returns), so a failure listener on another thread, e.g. a kill thread,
must not wait for
+ // either. Nothing bounds a wait for a kill thread: with
`spark.task.reaper.enabled`, every
+ // kill runs on a reaper thread, which checks
`spark.task.reaper.killTimeout` only after its
+ // kill call returns. If another thread holds the invocation and the task
is killed, the reaper
+ // with a positive kill timeout bounds the wait by killing the executor
JVM, except in local
+ // mode, where it only logs an error.
+ val currentThread = Thread.currentThread()
+ var mode: InvocationMode = Claimed
+ var awaitHandover = false
+ // Whether this call registered this thread as the listener invocation
thread or as the next
+ // one, so that it must make sure not to leave it registered.
+ var registered = false
val preCompletionExceptions = new ArrayBuffer[Throwable](2)
def nextPreCompletionListener(): Option[TaskCompletionListener] =
synchronized {
if (onPreCompleteCallbacks.empty()) {
// Set together with finding no more pre-completion listeners, so that
one added later
- // is handed to `addTaskCompletionListener` instead of being left in
the stack.
+ // is handed to `addTaskCompletionListener` instead of being left in
the stack. The
+ // listener invocation is claimed in the same step (see above).
completed = true
+ if (listenerInvocationThread.isEmpty) {
+ listenerInvocationThread = Some(currentThread)
+ registered = true
+ } else if (listenerInvocationThread.contains(currentThread)) {
+ // Called from a listener on this thread: run the completion
listeners as part of that
+ // invocation.
+ mode = Nested
+ } else {
+ nextListenerInvocationThread = Some(currentThread)
+ awaitHandover = true
+ registered = true
+ }
None
} else {
Some(onPreCompleteCallbacks.pop())
}
}
var listener: Option[TaskCompletionListener] = None
- while ({listener = nextPreCompletionListener(); listener.nonEmpty}) {
- invokeTaskPreCompletionListener(listener.get, preCompletionExceptions)
- }
-
try {
- invokeTaskCompletionListeners(error)
- } catch {
- case e: TaskCompletionListenerException if
preCompletionExceptions.nonEmpty =>
- // Merge the completion listeners' own failures, not the exception
that wraps them. Any
- // other error comes from invokeListeners itself (e.g. an
OutOfMemoryError) and is left
- // to propagate, so that the executor still sees a fatal one.
- preCompletionExceptions ++= e.getSuppressed.toSeq
+ while ({listener = nextPreCompletionListener(); listener.nonEmpty}) {
+ invokeTaskPreCompletionListener(listener.get, preCompletionExceptions)
+ }
+ try {
+ val warningFailure =
+ if (awaitHandover) awaitListenerInvocationHandover(currentThread)
else None
+ try {
+ invokeTaskCompletionListeners(error, mode)
+ } catch {
+ // The failure to warn came after the pre-completion listeners'
failures, but before
+ // those of the completion listeners.
+ case e: Throwable if warningFailure.exists(_ ne e) =>
warningFailure.get.addSuppressed(e)
Review Comment:
**2. [Low] A failure to warn hides a fatal error from the completion
listener invocation**
This catch takes any `Throwable`. Suppose the warning failed with a
non-fatal error, e.g. an `AppenderLoggingException`, and the invocation at L308
then throws an `OutOfMemoryError`. The `OutOfMemoryError` becomes a suppressed
exception of the failure to warn, and L314 throws the non-fatal one.
`Executor.isFatalError` follows causes, not suppressed exceptions
(`Executor.scala:1654-1665`), so the executor keeps running. L318-319 say that
a fatal error propagates.
Could this take only what is not fatal, with the same test as in item 1?
```scala
case e if !Utils.isFatalError(e) && warningFailure.exists(_ ne e)
=>
warningFailure.get.addSuppressed(e)
```
The fatal error then propagates instead of the failure to warn, which
matters less.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -288,27 +383,123 @@ private[spark] class TaskContextImpl(
exception
}
- private def invokeTaskCompletionListeners(error: Option[Throwable]): Unit = {
+ /**
+ * Waits until the thread invoking listeners hands the invocation over to
`currentThread`, which
+ * has registered itself as `nextListenerInvocationThread`.
+ *
+ * Interrupts do not stop the wait, and are passed on once it is done. The
wait of
+ * `PipelinedWriterTask.awaitExit()` in `BasePythonRunner` drops them
instead, because it is a
+ * listener itself and the listeners right after it must still be able to
clean up. This wait
+ * is not a listener but part of `markTaskCompleted()`, so it leaves the
interrupt status (e.g.
Review Comment:
**6. [Nit] With a kill thread as holder, the interrupt comes right after the
handover**
L393-394 say that this wait leaves the interrupt status as the completion
listeners would see it without the wait. When the holder is a kill thread, that
kill's interrupt does not arrive during this wait. `Task.kill` calls
`taskThread.interrupt()` only after `markInterrupted` has returned
(`Task.scala:237-239`), and by then L475 has handed the invocation over. So
with `interruptThread = true`, the completion listeners on this thread start at
about the same moment as the interrupt, and the flag can stay set for the ones
after it. `PipelinedWriterTask.awaitExit` drops interrupts to protect these
listeners, because the flag makes their blocking channel and lock operations
fail and skip cleanup (`PythonRunner.scala:378-381`).
This is not a regression, since master skips these listeners in this case,
and I don't see a simple way to avoid it here. But it is the main path of this
PR, so it may be worth a sentence here and in the description's note on the
wait, so that nobody assumes these listeners run undisturbed.
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -1195,19 +1192,570 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
completion
}
+ /** A daemon thread that records what `body` throws, for `joinThread` to
check. */
+ private class DaemonThread(name: String, body: => Unit) extends Thread(name)
{
+ setDaemon(true)
+ @volatile var error: Throwable = _
+ override def run(): Unit = {
+ try {
+ body
+ } catch {
+ case t: Throwable => error = t
+ }
+ }
+ }
+
+ private def startDaemonThread(name: String)(body: => Unit): DaemonThread = {
+ val thread = new DaemonThread(name, body)
+ thread.start()
+ thread
+ }
+
+ private def joinThread(thread: DaemonThread): Unit = {
+ thread.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!thread.isAlive, s"${thread.getName} is still running")
+ Option(thread.error).foreach(e => fail(s"${thread.getName} failed", e))
+ }
+
+ /**
+ * Runs `takeInvocation` on a new thread named `name`, by default failing
the task, and waits
+ * until a failure listener on it holds the listener invocation, which it
does until the
+ * returned latch is counted down. Returns that latch and the thread.
+ */
+ private def holdListenerInvocation(context: TaskContext, name: String =
"listener-holder")(
+ takeInvocation: => Unit = context.markTaskFailed(new Exception("failed
on another thread")))
+ : (CountDownLatch, DaemonThread) = {
+ val holding = new CountDownLatch(1)
+ val release = new CountDownLatch(1)
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= {
+ holding.countDown()
+ release.await(1, TimeUnit.MINUTES)
+ }
+ })
+ val holder = startDaemonThread(name)(takeInvocation)
+ assert(holding.await(1, TimeUnit.MINUTES))
+ (release, holder)
+ }
+
/**
- * 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.
+ * Fails every logging call whose message contains `trigger` with `failure`.
Unlike
+ * `LogAppender`, it does not ignore its exceptions, so log4j passes them to
the caller.
*/
private class FailingAppender(trigger: String, failure: RuntimeException)
extends AbstractAppender("failingAppender", null, null, false,
Property.EMPTY_ARRAY) {
+ val failures = new AtomicInteger()
+ // Of these, the failures of events that carry a throwable.
+ val failuresWithThrown = new AtomicInteger()
override def append(event: LogEvent): Unit = {
if (event.getMessage.getFormattedMessage.contains(trigger)) {
+ failures.incrementAndGet()
+ if (event.getThrown != null) failuresWithThrown.incrementAndGet()
throw failure
}
}
}
+ private def isCausedBy(error: Throwable, cause: Throwable): Boolean =
+ Iterator.iterate(error)(_.getCause).takeWhile(_ != null).exists(_ eq cause)
+
+ /** An exception whose `getMessage`, which is listener code too, throws
`failure`. */
+ private class MessageFailingException(failure: Throwable) extends Exception {
+ override def getMessage: String = throw failure
+ }
+
+ private def addFailureListenerThrowing(context: TaskContext)(failure: =>
Throwable): Unit =
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= throw failure
+ })
+
+ /** Waits until `thread` waits in `markTaskCompleted()` for the listener
invocation. */
+ private def awaitHandoverWait(thread: Thread): Unit = {
+ eventually(timeout(1.minute)) {
+ assert(!thread.isAlive || thread.getState == Thread.State.TIMED_WAITING)
+ }
+ assert(thread.isAlive, s"${thread.getName} returned instead of waiting")
+ }
+
+ test("SPARK-60222: a listener added as completion starts does not take over
the listeners") {
+ // A thread that sees the task completed and adds a listener at that
moment must not become
+ // the thread that runs the pending listeners: the completing thread
claims the listener
+ // invocation together with setting `completed`.
+ (1 to 500).foreach { _ =>
+ val context = TaskContext.empty()
+ val pendingListenerThread = new AtomicReference[Thread]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ pendingListenerThread.set(Thread.currentThread())
+ }
+ val ready = new CountDownLatch(1)
+ val adder = startDaemonThread("adder") {
+ ready.countDown()
+ while (!context.isCompleted()) {}
+ context.addTaskCompletionListener[Unit] { _ => }
+ }
+ ready.await()
+ context.markTaskCompleted(None)
+ joinThread(adder)
+ assert(pendingListenerThread.get eq Thread.currentThread())
+ }
+ }
+
+ test("SPARK-60222: a failing listener does not let another thread take over
the listeners") {
+ // While a failing listener's failure listeners run, the thread invoking
the listeners must
+ // stay registered, so that another thread adding listeners at that moment
does not start
+ // running the pending ones, possibly concurrently with this thread.
+ (1 to 200).foreach { _ =>
+ val context = TaskContext.empty()
+ val pendingListenerThread = new AtomicReference[Thread]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ pendingListenerThread.set(Thread.currentThread())
+ }
+ // Two failing listeners, which run first: the second failure finds the
task already failed.
+ (1 to 2).foreach { i =>
+ context.addTaskCompletionListener(new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw
new Exception(s"$i")
+ })
+ }
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable):
Unit = {}
+ })
+ val ready = new CountDownLatch(1)
+ val adder = startDaemonThread("adder") {
+ ready.countDown()
+ while (!context.isCompleted()) {}
+ (1 to 200).foreach { _ => context.addTaskCompletionListener[Unit] { _
=> } }
+ }
+ ready.await()
+ intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ joinThread(adder)
+ assert(pendingListenerThread.get eq Thread.currentThread())
+ }
+ }
+
+ test("SPARK-60222: completion waits for another thread invoking listeners
and runs them") {
+ val context = TaskContext.empty()
+ val listenerThreads = new ConcurrentLinkedQueue[(String, Thread)]()
+ context.addTaskCompletionListener(loggingListener(listenerThreads,
"pending"))
+ context.addTaskInterruptListener(new TaskInterruptListener {
+ override def onTaskInterrupted(context: TaskContext, reason: String):
Unit =
+ throw new Exception("interrupt listener")
+ })
+ // Like a kill: an interrupt listener fails, so the kill thread runs the
failure listeners.
+ val (release, killer) = holdListenerInvocation(context, "killer") {
+ context.markInterrupted("killed")
+ }
+
+ val completion =
startDaemonThread("completion")(context.markTaskCompleted(None))
+ try {
+ awaitHandoverWait(completion)
+ // Neither returns early nor runs the completion listeners while the
kill thread holds the
+ // listener invocation, and a third thread adding a listener does not
take them over.
+ joinThread(startDaemonThread("adder") {
+ context.addTaskCompletionListener(loggingListener(listenerThreads,
"added"))
+ })
+ assert(completion.isAlive)
+ assert(listenerThreads.isEmpty)
+ } finally {
+ release.countDown()
+ }
+ joinThread(completion)
+ joinThread(killer)
+ assert(listenerThreads.asScala.toSeq === Seq(("added", completion),
("pending", completion)))
+ }
+
+ test("SPARK-60222: a listener on the invoking thread completing the task
runs them nested") {
+ val context = TaskContext.empty()
+ val completionListenerThread = new AtomicReference[Thread]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ completionListenerThread.set(Thread.currentThread())
+ }
+ val ranBeforeReturn = new AtomicReference[Option[Boolean]](None)
+ val laterListenerThread = new AtomicReference[Thread]()
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= {
+ // This thread holds the listener invocation, so it must not wait for
itself.
+ context.markTaskCompleted(Some(error))
+ ranBeforeReturn.set(Some(completionListenerThread.get eq
Thread.currentThread()))
+ // And it still holds the invocation, so another thread adding a
listener does not run it.
+ joinThread(startDaemonThread("adder") {
+ context.addTaskCompletionListener[Unit] { _ =>
+ laterListenerThread.set(Thread.currentThread())
+ }
+ })
+ }
+ })
+ joinThread(startDaemonThread("failer")(context.markTaskFailed(new
Exception("failed"))))
+ assert(ranBeforeReturn.get === Some(true))
+ assert(laterListenerThread.get == null)
+ }
+
+ test("SPARK-60222: a failure listener left to the completing thread runs
before the next " +
+ "completion listener") {
+ val context = TaskContext.empty()
+ val invocations = new ConcurrentLinkedQueue[(String, Thread)]()
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable): Unit
= {
+ invocations.add(("failure", Thread.currentThread()))
+ throw new Exception("failure listener")
+ }
+ })
+ // E.g. a completion listener that releases a state store, which a failure
listener aborts.
+ context.addTaskCompletionListener(loggingListener(invocations, "second"))
+ context.addTaskCompletionListener[Unit] { _ =>
+ invocations.add(("first", Thread.currentThread()))
+ // Another thread fails the task while this thread runs the completion
listeners, e.g. a
+ // kill whose interrupt listener failed. It finds the listener
invocation taken and leaves
+ // the failure listeners to this thread.
+ joinThread(startDaemonThread("failer") {
+ context.markTaskFailed(new Exception("failed elsewhere"))
+ })
+ assert(invocations.size == 1)
+ }
+
+ val error = intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ val current = Thread.currentThread()
+ assert(invocations.asScala.toSeq ===
+ Seq(("first", current), ("failure", current), ("second", current)))
+ // The failure listener's failure is reported as its own, not wrapped in
the exception of the
+ // failure listener invocation.
+ assert(error.getSuppressed.map(_.getMessage).toSeq === Seq("failure
listener"))
+ }
+
+ test("SPARK-60222: a failure listener left to the thread the invocation was
handed over to " +
+ "runs before its next completion listener") {
+ val context = TaskContext.empty()
+ val invocations = new ConcurrentLinkedQueue[(String, Thread)]()
+ context.addTaskCompletionListener(loggingListener(invocations, "second"))
+ context.addTaskCompletionListener[Unit] { _ =>
+ invocations.add(("first", Thread.currentThread()))
+ // The task has failed, so the added failure listener is to run right
away. This thread
+ // holds the listener invocation, so the adding thread leaves it to this
one.
+ joinThread(startDaemonThread("adder") {
+ context.addTaskFailureListener(new TaskFailureListener {
+ override def onTaskFailure(context: TaskContext, error: Throwable):
Unit =
+ invocations.add(("failure", Thread.currentThread()))
+ })
+ })
+ assert(invocations.size == 1)
+ }
+ val (release, holder) = holdListenerInvocation(context)()
+ val completion =
startDaemonThread("completion")(context.markTaskCompleted(None))
+ try {
+ awaitHandoverWait(completion)
+ } finally {
+ release.countDown()
+ }
+ joinThread(completion)
+ joinThread(holder)
+ assert(invocations.asScala.toSeq ===
+ Seq(("first", completion), ("failure", completion), ("second",
completion)))
+ }
+
+ test("SPARK-60222: an exception whose getMessage throws does not end the
listener loops") {
+ val messageFailure = new IllegalStateException("no message")
+ val current = Thread.currentThread()
+ // The failure listener invocation that a failing pre-completion or
completion listener
+ // starts throws the failure of `getMessage`, while building its exception.
+ Seq(true, false).foreach { preCompletion =>
+ val context = TaskContext.empty()
+ val invocations = new ConcurrentLinkedQueue[(String, Thread)]()
+ addFailureListenerThrowing(context)(new
MessageFailingException(messageFailure))
+ context.addTaskCompletionListener(loggingListener(invocations, "next"))
+ val failing = new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw new
Exception("listener")
+ }
+ if (preCompletion) {
+ context.addTaskPreCompletionListener(failing)
+ } else {
+ context.addTaskCompletionListener(failing)
+ }
+ val error = intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ assert(invocations.asScala.toSeq === Seq(("next", current)),
s"preCompletion=$preCompletion")
+ assert(context.isCompleted())
+ val listenerFailure = error.getSuppressed.toSeq.head
+ assert(listenerFailure.getMessage === "listener")
+ assert(listenerFailure.getSuppressed.exists(_ eq messageFailure))
+ }
+
+ // Likewise for the failure listeners left to the completing thread.
+ val context = TaskContext.empty()
+ val invocations = new ConcurrentLinkedQueue[(String, Thread)]()
+ addFailureListenerThrowing(context)(new
MessageFailingException(messageFailure))
+ context.addTaskCompletionListener(loggingListener(invocations, "second"))
+ context.addTaskCompletionListener[Unit] { _ =>
+ joinThread(startDaemonThread("failer") {
+ context.markTaskFailed(new Exception("failed elsewhere"))
+ })
+ }
+ val error = intercept[TaskCompletionListenerException] {
+ context.markTaskCompleted(None)
+ }
+ assert(invocations.asScala.toSeq === Seq(("second", current)))
+ assert(error.getSuppressed.toSeq === Seq(messageFailure))
+ }
+
+ test("SPARK-60222: an error that escapes the listener loop still releases
the invocation") {
+ val context = TaskContext.empty()
+ context.markTaskCompleted(None)
+ val fatal = new LinkageError("no message")
+ addFailureListenerThrowing(context)(new MessageFailingException(fatal))
+ val error = intercept[LinkageError] {
+ // Runs right away on this thread and fails, and the failure listener
invocation that this
+ // starts throws `fatal` while building its exception.
+ context.addTaskCompletionListener[Unit] { _ => throw new
Exception("listener") }
+ }
+ assert(error eq fatal)
+ // Released: a listener added now runs right away on this thread.
+ val laterListenerThread = new AtomicReference[Thread]()
+ context.addTaskCompletionListener[Unit] { _ =>
+ laterListenerThread.set(Thread.currentThread())
+ }
+ assert(laterListenerThread.get eq Thread.currentThread())
+ }
+
+ test("SPARK-60222: a fatal error of a failure listener invocation is not
hidden by the failure " +
+ "of the listener that started it") {
+ val fatal = new LinkageError("no message")
+ // Both for a failing completion listener and for a failing pre-completion
listener.
+ Seq(true, false).foreach { preCompletion =>
+ val context = TaskContext.empty()
+ addFailureListenerThrowing(context)(new MessageFailingException(fatal))
+ val failing = new TaskCompletionListener {
+ override def onTaskCompletion(context: TaskContext): Unit = throw new
Exception("listener")
+ }
+ if (preCompletion) {
+ context.addTaskPreCompletionListener(failing)
+ } else {
+ context.addTaskCompletionListener(failing)
+ }
+ // It propagates, so that the executor sees it, instead of being a
suppressed exception of
+ // the listener's failure.
+ val error = intercept[LinkageError] {
+ context.markTaskCompleted(None)
+ }
+ assert(error eq fatal, s"preCompletion=$preCompletion")
+ }
+ }
+
+ test("SPARK-60222: an interrupt does not stop the handover wait and is
kept") {
+ val context = TaskContext.empty()
+ val listenerThreads = new ConcurrentLinkedQueue[(String, Thread)]()
+ (1 to 2).foreach { i =>
+ context.addTaskCompletionListener(loggingListener(listenerThreads,
s"completion $i"))
+ }
+ val (release, holder) = holdListenerInvocation(context)()
+ val interruptedAfterCompletion = new AtomicReference[Option[Boolean]](None)
+ val completion = startDaemonThread("completion") {
+ context.markTaskCompleted(None)
+
interruptedAfterCompletion.set(Some(Thread.currentThread().isInterrupted))
+ }
+ try {
+ awaitHandoverWait(completion)
+ completion.interrupt()
+ // Still waiting for the holder, without having run any completion
listener.
+ eventually(timeout(1.minute)) {
+ assert(!completion.isAlive ||
+ !completion.isInterrupted && completion.getState ==
Thread.State.TIMED_WAITING)
+ }
+ assert(completion.isAlive, "completion returned instead of waiting")
+ assert(listenerThreads.isEmpty)
+ } finally {
+ release.countDown()
+ }
+ joinThread(completion)
+ joinThread(holder)
+ assert(interruptedAfterCompletion.get === Some(true))
+ assert(listenerThreads.asScala.toSeq ===
+ Seq(("completion 2", completion), ("completion 1", completion)))
+ }
+
+ test("SPARK-60222: the handover wait warns with the holder's name and stack
trace") {
+ val context = TaskContext.empty()
+ context.listenerHandoverWarnIntervalMs = 100
+ val (release, holder) = holdListenerInvocation(context)()
+ val logAppender = new LogAppender("listener handover wait")
+ var completion: DaemonThread = null
+ try {
+ withLogAppender(logAppender, Seq(classOf[TaskContextImpl].getName),
Some(Level.WARN)) {
+ completion =
startDaemonThread("completion")(context.markTaskCompleted(None))
+ eventually(timeout(1.minute)) {
+ assert(!completion.isAlive ||
+ logAppender.loggingEvents.count(_.getLevel == Level.WARN) >= 2)
+ }
+ assert(completion.isAlive, "completion returned instead of waiting")
+ }
+ } finally {
+ release.countDown()
+ }
+ joinThread(completion)
+ joinThread(holder)
+ val warnings = logAppender.loggingEvents.filter(_.getLevel ==
Level.WARN).toSeq
+ warnings.foreach { event =>
+ assert(event.getMessage.getFormattedMessage.contains("until thread
listener-holder"))
+ }
+ // The first warning carries the holder's stack trace, blocked in its
failure listener.
+ val holderStack = warnings.head.getThrown
+ assert(holderStack.getMessage.contains("listener-holder"))
+ assert(holderStack.getStackTrace.exists(_.getMethodName ==
"onTaskFailure"))
+ // The later ones do not walk the stack again.
+ warnings.tail.foreach { event => assert(event.getThrown == null) }
+ }
+
+ test("SPARK-60222: a failing handover warning does not end the wait") {
Review Comment:
**4. [Nit] No test covers a failure to warn together with a failing
completion listener**
`TaskContextImpl.scala:312` adds the completion listeners' failure to the
failure to warn. Neither test reaches it: the `pending` listener here does not
fail, and "a failing handover warning is reported with the failures of the
pre-completion listeners" has no completion listener. Without that catch both
still pass, although the `TaskCompletionListenerException` would then skip
`TaskContextImpl.scala:314` and the failure to warn would be lost. The
description's list of what is not covered does not mention this.
Could this test get a failing completion listener and check that
`completionError` carries both failures? If item 2 changes that catch, an
`OutOfMemoryError` variant would cover that too.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -266,7 +358,10 @@ private[spark] class TaskContextImpl(
try {
markTaskFailed(e)
} catch {
- case t: Throwable => e.addSuppressed(t)
+ // Failures of the failure listeners, or e.g. of the `getMessage` of
one of their
+ // exceptions, which is listener code too. A fatal error (e.g. an
OutOfMemoryError)
+ // propagates, so that the executor still sees it.
+ case NonFatal(t) => e.addSuppressed(t)
Review Comment:
**1. [Low] `NonFatal` lets a `LinkageError` end the listener loops, while
the executor keeps running**
This corrects my own suggestion. In [round-3 item
1](https://github.com/apache/spark/pull/59663#discussion_r4419200062) I
proposed `NonFatal`, and in [round-3 item
7](https://github.com/apache/spark/pull/59663#discussion_r4419200095) I named a
`LinkageError` as an error that still propagates. `NonFatal` does not match a
`LinkageError`, but the executor does not count it as fatal either
(`Utils.scala:1976-1987`, `Executor.scala:1654-1665`). So propagating it does
not stop the executor, and it costs the listeners. On master, `case t:
Throwable => e.addSuppressed(t)` kept these loops going.
Take a task whose body succeeded. A pre-completion listener fails, and a
failure listener throws an exception whose `getMessage` throws a
`LinkageError`, e.g. a `NoClassDefFoundError`. "a fatal error of a failure
listener invocation is not hidden by the failure of the listener that started
it" does exactly this (`TaskContextSuite.scala:1522-1544`). Then:
- The error passes this catch and ends the loop at L301-303 before L281 sets
`completed`. `e` is neither recorded nor logged.
- No completion listener runs, so their cleanup is skipped, e.g. the release
of the state store stamp at `RocksDBStateStoreProvider.scala:189`.
- `completed` stays false, so a listener added later is only queued.
- The executor reports the failure and keeps running.
The catches at L644 and L666 skip the remaining completion listeners in the
same way. The test checks only which error comes out, not `isCompleted()` or
the listeners.
Could this catch use the executor's own test, and L322, L644 and L666 too?
Then only an error that stops the executor ends the loops:
```scala
case t if !Utils.isFatalError(t) => e.addSuppressed(t)
```
The two tests that use a `LinkageError` would then need an error such as an
`OutOfMemoryError`.
Also, when the task body failed, `runTaskWithListeners` turns whatever
`markTaskCompleted(Some(e))` throws into a suppressed exception of `e`
(`TaskContext.scala:250-255`). So "so that the executor still sees it" at
L318-319 holds only when the body succeeded.
##########
core/src/main/scala/org/apache/spark/TaskContextImpl.scala:
##########
@@ -346,10 +537,31 @@ private[spark] class TaskContextImpl(
}
}
+ /**
+ * Invokes and removes the listeners in `listeners`, in reverse order of
registration.
+ *
+ * Other threads cannot invoke listeners while this thread does, so they
leave the listeners
+ * they add, or whose invocation they start, meanwhile. Those in `listeners`
run here like the
+ * others. If this is the completion listener invocation of
`markTaskCompleted()`, the failure
+ * listeners left to it (the task has failed) run before the completion
listeners that have
+ * not run yet, e.g. so that a failure listener that aborts a state store
update runs before
+ * the completion listener that releases the state store, if that one has
not run yet. This is
+ * the order of a failure listener invocation that a failing completion
listener starts, and,
+ * likewise, the failures of these failure listeners are reported with those
of the completion
+ * listeners. Any other invocation does not run the listeners of another
kind left to it, since
Review Comment:
**5. [Nit] "must not run" does not hold for the failure listeners that a
failing listener starts**
L551-553 say that any other invocation does not run the listeners left to
it, since its thread may be one on which they must not run, e.g. a kill thread,
or the task thread after `Task.run`. But when a listener in such an invocation
throws, L639 runs the failure listeners on that very thread. A
post-status-update listener that throws does this after the executor has
released the task's block locks and memory, and so does a completion listener
that fails on a kill thread that added it after completion.
Master does the same, and I am not asking to change that. This is only about
the wording, which [your reply on round-3 item
6](https://github.com/apache/spark/pull/59663#discussion_r4421488978) relies
on. Could the sentence say that it is about listeners that other threads leave,
and that a failing listener on this thread still starts the failure listeners
here (see `runListener`)?
--
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]
