Skip to content
Closed
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -432,7 +432,7 @@ private[spark] class Executor(
setTaskFinishedAndClearInterruptStatus()
execBackend.statusUpdate(taskId, TaskState.KILLED, ser.serialize(TaskKilled(t.reason)))

case _: InterruptedException if task.reasonIfKilled.isDefined =>
case _: Throwable if task.reasonIfKilled.isDefined =>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Throwable -> NonFatal

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@mridulm, that was my initial reaction as well, but I noticed that we're already catching Throwable down on line 447.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That said, the existing Throwable case has logic to exit the JVM in case the throwable was fatal (see Utils.isFatalError(t)) used further down).

We probably want to preserve the behavior of continuing to exit on fatal errors while still trying to issue a final "dying breath" task status. Thus maybe we can catch NonFatal up here, as you've proposed, and add another piece of conditional logic to handle fatal errors in the existing Throwable case. This adds a bit of complexity and we'd need another test case to fully cover it, but that could be done by having a task which throws OutOfMemoryError once interrupted.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think NonFatal is probably fine here -- an OOM could presumably cause these messages to be dropped and the task marked as failed anyways.

val killReason = task.reasonIfKilled.getOrElse("unknown reason")
logInfo(s"Executor interrupted and killed $taskName (TID $taskId), reason: $killReason")
setTaskFinishedAndClearInterruptStatus()
Expand Down
8 changes: 7 additions & 1 deletion core/src/test/scala/org/apache/spark/SparkContextSuite.scala
Original file line number Diff line number Diff line change
Expand Up @@ -572,7 +572,13 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
// first attempt will hang
if (!SparkContextSuite.isTaskStarted) {
SparkContextSuite.isTaskStarted = true
Thread.sleep(9999999)
try {
Thread.sleep(9999999)
} catch {
case t: Throwable =>
// SPARK-20217 should not fail stage if task throws non-interrupted exception
throw new RuntimeException("killed")
}
}
// second attempt succeeds immediately
}
Expand Down