viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4428913065
##########
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:
Thanks for following up on this. These four catches now take what
`Utils.isFatalError` does not count as fatal, so a `LinkageError` is reported
with the listener's failure, and only an error that also ends the executor ends
the loops. With item 3 a failing `getMessage` no longer escapes the failure
listener invocation, but a `getMessage` whose failure fails in `getMessage` too
still does, so "an error that is not fatal to the executor does not end the
listener loops" uses that to throw a `LinkageError` after a failing
pre-completion or completion listener and from a left failure listener, and
checks that the next completion listener runs and the task is completed. It
also throws one from the completion listener invocation after a failing
pre-completion listener, which is then reported with that listener's failure
(the catch at L323). The two `LinkageError` tests now use an
`OutOfMemoryError`. The comments no longer say that the executor sees the
error; they say that it ends the ta
sk's listeners, as it ends the executor, and the test says the executor sees
it if the task body succeeded.
##########
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:
Done, as you suggested. "a failing handover warning is reported with the
failures of the completion listeners, unless one of them is fatal" covers the
case where the completion listener invocation throws an `OutOfMemoryError`.
##########
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:
Done: `newTaskCompletionListenerException` now reports the failure of
`getMessage` in place of the exception it fails for, and the interrupt listener
and `invokeListeners()` paths use it too. It calls `getMessage` once per
exception, collecting the message with the exception it reports. "a failure of
getMessage is reported in place of the exception it fails for" replaces the
test you mentioned, and checks your example with two left failure listeners,
two pre-completion listeners, and two interrupt listeners, where the failure is
now recorded instead of escaping `markInterrupted`.
##########
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:
Added "a failing handover warning is reported with the failures of the
completion listeners, unless one of them is fatal": the non-fatal variant
checks that the failure to warn carries the `TaskCompletionListenerException`
with the completion listener's failure, and fails without the catch at L312;
the other one is the `OutOfMemoryError` variant from item 2.
##########
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:
Reworded: the sentence is now about the listeners of another kind that other
threads left to the invocation, and says that a listener that fails there still
starts the failure listeners on that thread (see `runListener`).
##########
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:
Good point. Added a sentence to the scaladoc of
`awaitListenerInvocationHandover` and to the note on the wait in the
description: with a kill thread as holder, its interrupt comes once
`markInterrupted()` has returned, after the handover, so with `interruptThread`
it may come while the completion listeners run on the task thread.
--
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]
