viirya commented on code in PR #59663:
URL: https://github.com/apache/spark/pull/59663#discussion_r4404167897
##########
core/src/test/scala/org/apache/spark/scheduler/TaskContextSuite.scala:
##########
@@ -931,6 +931,466 @@ class TaskContextSuite extends SparkFunSuite with
BeforeAndAfter with LocalSpark
completion
}
+ private def startDaemonThread(name: String)(body: => Unit): Thread = {
+ val thread = new Thread(() => body, name)
+ thread.setDaemon(true)
+ thread.start()
+ thread
+ }
+
+ private def joinThread(thread: Thread): Unit = {
+ thread.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!thread.isAlive, s"${thread.getName} is still running")
+ }
+
+ /**
+ * 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, Thread) = {
+ 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)
+ }
+
+ /**
+ * 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) {
+ override def append(event: LogEvent): Unit = {
+ if (event.getMessage.getFormattedMessage.contains(trigger)) {
+ throw failure
+ }
+ }
+ }
+
+ private def isCausedBy(error: Throwable, cause: Throwable): Boolean =
+ Iterator.iterate(error)(_.getCause).takeWhile(_ != null).exists(_ eq cause)
+
+ /** Waits until `thread` waits in `markTaskCompleted()` for the listener
invocation. */
+ private def awaitHandoverWait(thread: Thread): Unit = {
+ eventually(timeout(1.minute)) {
Review Comment:
Done for all four: the polls also stop once the thread has ended, and an
assertion then checks that it is still alive.
--
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]
