[SPARK-20217][core] Executor should not fail stage if killed task throws non-interrupted exception #17531

ericl · 2017-04-05T00:09:08Z

What changes were proposed in this pull request?

If tasks throw non-interrupted exceptions on kill (e.g. java.nio.channels.ClosedByInterruptException), their death is reported back as TaskFailed instead of TaskKilled. This causes stage failure in some cases.

This is reproducible as follows. Run the following, and then use SparkContext.killTaskAttempt to kill one of the tasks. The entire stage will fail since we threw a RuntimeException instead of InterruptedException.

spark.range(100).repartition(100).foreach { i =>
  try {
    Thread.sleep(10000000)
  } catch {
    case t: InterruptedException =>
      throw new RuntimeException(t)
  }
}

Based on the code in TaskSetManager, I think this also affects kills of speculative tasks. However, since the number of speculated tasks is few, and usually you need to fail a task a few times before the stage is cancelled, it unlikely this would be noticed in production unless both speculation was enabled and the num allowed task failures was = 1.

We should probably unconditionally return TaskKilled instead of TaskFailed if the task was killed by the driver, regardless of the actual exception thrown.

How was this patch tested?

Unit test. The test fails before the change in Executor.scala

cc @JoshRosen

SparkQA · 2017-04-05T02:17:12Z

Test build #75525 has finished for PR 17531 at commit 9d59960.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

mridulm · 2017-04-05T09:10:21Z

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -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 =>


Throwable -> NonFatal

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

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.

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

mridulm · 2017-04-05T21:06:50Z

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -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 NonFatal(_) if task != null && task.reasonIfKilled.isDefined =>


ah, task != null, excellent catch; completely missed that !

mridulm

LGTM, thanks for the fix @ericl !

mridulm · 2017-04-05T21:07:59Z

I will leave it around a bit in case @JoshRosen has any further comments. Feel free to merge btw if you dont !

SparkQA · 2017-04-05T21:12:40Z

Test build #75552 has finished for PR 17531 at commit 28ebc94.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-04-05T21:18:12Z

Test build #75553 has finished for PR 17531 at commit 42f49f2.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

yhuai · 2017-04-05T21:31:25Z

test this please

JoshRosen · 2017-04-05T21:49:44Z

LGTM pending Jenkins.

SparkQA · 2017-04-06T00:23:59Z

Test build #75559 has finished for PR 17531 at commit 42f49f2.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

yhuai · 2017-04-06T00:28:57Z

Thanks. Merging to master.

…thrown by cancelled tasks ## What changes were proposed in this pull request? This was a regression introduced by my earlier PR here: #17531 It turns out NonFatal() does not in fact catch InterruptedException. ## How was this patch tested? Extended cancellation unit test coverage. The first test fails before this patch. cc JoshRosen mridulm Author: Eric Liang <ekl@databricks.com> Closes #17659 from ericl/spark-20358. (cherry picked from commit b2ebadf) Signed-off-by: Yin Huai <yhuai@databricks.com>

…thrown by cancelled tasks ## What changes were proposed in this pull request? This was a regression introduced by my earlier PR here: apache#17531 It turns out NonFatal() does not in fact catch InterruptedException. ## How was this patch tested? Extended cancellation unit test coverage. The first test fails before this patch. cc JoshRosen mridulm Author: Eric Liang <ekl@databricks.com> Closes apache#17659 from ericl/spark-20358.

…ows non-interrupted exception ## What changes were proposed in this pull request? If tasks throw non-interrupted exceptions on kill (e.g. java.nio.channels.ClosedByInterruptException), their death is reported back as TaskFailed instead of TaskKilled. This causes stage failure in some cases. This is reproducible as follows. Run the following, and then use SparkContext.killTaskAttempt to kill one of the tasks. The entire stage will fail since we threw a RuntimeException instead of InterruptedException. ``` spark.range(100).repartition(100).foreach { i => try { Thread.sleep(10000000) } catch { case t: InterruptedException => throw new RuntimeException(t) } } ``` Based on the code in TaskSetManager, I think this also affects kills of speculative tasks. However, since the number of speculated tasks is few, and usually you need to fail a task a few times before the stage is cancelled, it unlikely this would be noticed in production unless both speculation was enabled and the num allowed task failures was = 1. We should probably unconditionally return TaskKilled instead of TaskFailed if the task was killed by the driver, regardless of the actual exception thrown. ## How was this patch tested? Unit test. The test fails before the change in Executor.scala cc JoshRosen Author: Eric Liang <ekl@databricks.com> Closes apache#17531 from ericl/fix-task-interrupt.

…thrown by cancelled tasks This was a regression introduced by my earlier PR here: apache#17531 It turns out NonFatal() does not in fact catch InterruptedException. Extended cancellation unit test coverage. The first test fails before this patch. cc JoshRosen mridulm Author: Eric Liang <ekl@databricks.com> Closes apache#17659 from ericl/spark-20358.

ericl added 2 commits April 4, 2017 16:52

Tue Apr 4 16:52:51 PDT 2017

8f6283a

Tue Apr 4 17:04:06 PDT 2017

9d59960

mridulm reviewed Apr 5, 2017

View reviewed changes

ericl added 2 commits April 5, 2017 11:50

Wed Apr 5 11:50:42 PDT 2017

28ebc94

add null check for task

42f49f2

mridulm reviewed Apr 5, 2017

View reviewed changes

mridulm approved these changes Apr 5, 2017

View reviewed changes

asfgit closed this in 5142e5d Apr 6, 2017

ericl mentioned this pull request Apr 17, 2017

[SPARK-20358] [core] Executors failing stage on interrupted exception thrown by cancelled tasks #17659

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-20217][core] Executor should not fail stage if killed task throws non-interrupted exception #17531

[SPARK-20217][core] Executor should not fail stage if killed task throws non-interrupted exception #17531

ericl commented Apr 5, 2017 •

edited

Loading

SparkQA commented Apr 5, 2017

mridulm Apr 5, 2017

JoshRosen Apr 5, 2017

JoshRosen Apr 5, 2017

ericl Apr 5, 2017

mridulm Apr 5, 2017

mridulm left a comment

mridulm commented Apr 5, 2017

SparkQA commented Apr 5, 2017

SparkQA commented Apr 5, 2017

yhuai commented Apr 5, 2017

JoshRosen commented Apr 5, 2017

SparkQA commented Apr 6, 2017

yhuai commented Apr 6, 2017

[SPARK-20217][core] Executor should not fail stage if killed task throws non-interrupted exception #17531

[SPARK-20217][core] Executor should not fail stage if killed task throws non-interrupted exception #17531

Conversation

ericl commented Apr 5, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Apr 5, 2017

mridulm Apr 5, 2017

Choose a reason for hiding this comment

JoshRosen Apr 5, 2017

Choose a reason for hiding this comment

JoshRosen Apr 5, 2017

Choose a reason for hiding this comment

ericl Apr 5, 2017

Choose a reason for hiding this comment

mridulm Apr 5, 2017

Choose a reason for hiding this comment

mridulm left a comment

Choose a reason for hiding this comment

mridulm commented Apr 5, 2017

SparkQA commented Apr 5, 2017

SparkQA commented Apr 5, 2017

yhuai commented Apr 5, 2017

JoshRosen commented Apr 5, 2017

SparkQA commented Apr 6, 2017

yhuai commented Apr 6, 2017

ericl commented Apr 5, 2017 •

edited

Loading