[SPARK-19276][CORE] Fetch Failure handling robust to user error handling #16639
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,6 +18,7 @@ | |
| package org.apache.spark.executor | ||
|
|
||
| import java.io.{Externalizable, ObjectInput, ObjectOutput} | ||
| import java.lang.Thread.UncaughtExceptionHandler | ||
| import java.nio.ByteBuffer | ||
| import java.util.Properties | ||
| import java.util.concurrent.{CountDownLatch, TimeUnit} | ||
|
|
@@ -27,7 +28,7 @@ import scala.concurrent.duration._ | |
|
|
||
| import org.mockito.ArgumentCaptor | ||
| import org.mockito.Matchers.{any, eq => meq} | ||
| import org.mockito.Mockito.{inOrder, verify, when} | ||
| import org.mockito.invocation.InvocationOnMock | ||
| import org.mockito.stubbing.Answer | ||
| import org.scalatest.concurrent.Eventually | ||
|
|
@@ -37,9 +38,12 @@ import org.apache.spark._ | |
| import org.apache.spark.TaskState.TaskState | ||
| import org.apache.spark.memory.MemoryManager | ||
| import org.apache.spark.metrics.MetricsSystem | ||
| import org.apache.spark.rdd.RDD | ||
| import org.apache.spark.rpc.RpcEnv | ||
| import org.apache.spark.scheduler.{FakeTask, ResultTask, TaskDescription} | ||
| import org.apache.spark.serializer.JavaSerializer | ||
| import org.apache.spark.shuffle.FetchFailedException | ||
| import org.apache.spark.storage.BlockManagerId | ||
|
|
||
| class ExecutorSuite extends SparkFunSuite with LocalSparkContext with MockitoSugar with Eventually { | ||
|
|
||
|
|
@@ -123,6 +127,75 @@ class ExecutorSuite extends SparkFunSuite with LocalSparkContext with MockitoSug | |
| } | ||
| } | ||
|
|
||
| test("SPARK-19276: Handle FetchFailedExceptions that are hidden by user exceptions") { | ||
| val conf = new SparkConf().setMaster("local").setAppName("executor suite test") | ||
| sc = new SparkContext(conf) | ||
| val serializer = SparkEnv.get.closureSerializer.newInstance() | ||
| val resultFunc = (context: TaskContext, itr: Iterator[Int]) => itr.size | ||
|
|
||
| // Submit a job where a fetch failure is thrown, but user code has a try/catch which hides | ||
| // the fetch failure. The executor should still tell the driver that the task failed due to a | ||
| // fetch failure, not a generic exception from user code. | ||
| val inputRDD = new FetchFailureThrowingRDD(sc) | ||
| val secondRDD = new FetchFailureHidingRDD(sc, inputRDD, throwOOM = false) | ||
| val taskBinary = sc.broadcast(serializer.serialize((secondRDD, resultFunc)).array()) | ||
| val serializedTaskMetrics = serializer.serialize(TaskMetrics.registered).array() | ||
| val task = new ResultTask( | ||
| stageId = 1, | ||
| stageAttemptId = 0, | ||
| taskBinary = taskBinary, | ||
| partition = secondRDD.partitions(0), | ||
| locs = Seq(), | ||
| outputId = 0, | ||
| localProperties = new Properties(), | ||
| serializedTaskMetrics = serializedTaskMetrics | ||
| ) | ||
|
|
||
| val serTask = serializer.serialize(task) | ||
| val taskDescription = createFakeTaskDescription(serTask) | ||
|
|
||
| val failReason = runTaskAndGetFailReason(taskDescription) | ||
|
Contributor
There was a problem hiding this comment. can you add a comment about what's going on here? I think the FFE gets thrown because the shuffle map data was never generated? And then you're checking that it's correctly accounted for, even though the user RDD code wrapped the exception in something else? |
||
| assert(failReason.isInstanceOf[FetchFailed]) | ||
| } | ||
|
|
||
| test("SPARK-19276: OOMs correctly handled with a FetchFailure") { | ||
| // when there is a fatal error like an OOM, we don't do normal fetch failure handling, since it | ||
| // may be a false positive. And we should call the uncaught exception handler. | ||
| val conf = new SparkConf().setMaster("local").setAppName("executor suite test") | ||
| sc = new SparkContext(conf) | ||
| val serializer = SparkEnv.get.closureSerializer.newInstance() | ||
| val resultFunc = (context: TaskContext, itr: Iterator[Int]) => itr.size | ||
|
|
||
| // Submit a job where a fetch failure is thrown, but user code has a try/catch which hides | ||
| // the fetch failure. The executor should still tell the driver that the task failed due to a | ||
| // fetch failure, not a generic exception from user code. | ||
| val inputRDD = new FetchFailureThrowingRDD(sc) | ||
| val secondRDD = new FetchFailureHidingRDD(sc, inputRDD, throwOOM = true) | ||
| val taskBinary = sc.broadcast(serializer.serialize((secondRDD, resultFunc)).array()) | ||
| val serializedTaskMetrics = serializer.serialize(TaskMetrics.registered).array() | ||
| val task = new ResultTask( | ||
| stageId = 1, | ||
| stageAttemptId = 0, | ||
| taskBinary = taskBinary, | ||
| partition = secondRDD.partitions(0), | ||
| locs = Seq(), | ||
| outputId = 0, | ||
| localProperties = new Properties(), | ||
| serializedTaskMetrics = serializedTaskMetrics | ||
| ) | ||
|
|
||
| val serTask = serializer.serialize(task) | ||
| val taskDescription = createFakeTaskDescription(serTask) | ||
|
|
||
| val (failReason, uncaughtExceptionHandler) = | ||
| runTaskGetFailReasonAndExceptionHandler(taskDescription) | ||
| assert(failReason.isInstanceOf[ExceptionFailure]) | ||
| val exceptionCaptor = ArgumentCaptor.forClass(classOf[Throwable]) | ||
| verify(uncaughtExceptionHandler).uncaughtException(any(), exceptionCaptor.capture()) | ||
| assert(exceptionCaptor.getAllValues.size === 1) | ||
| assert(exceptionCaptor.getAllValues.get(0).isInstanceOf[OutOfMemoryError]) | ||
| } | ||
|
|
||
| test("Gracefully handle error in task deserialization") { | ||
|
Contributor
There was a problem hiding this comment. is this test related to this PR? (seems useful but like it should be in its own PR?)
Contributor
Author
There was a problem hiding this comment. @mridulm pointed out this bug in an earlier version of this pr, so I fixed the bug and added a test case. But in any case, I've separated this out into #16930 / https://issues.apache.org/jira/browse/SPARK-19597 |
||
| val conf = new SparkConf | ||
| val serializer = new JavaSerializer(conf) | ||
|
|
@@ -169,13 +242,20 @@ class ExecutorSuite extends SparkFunSuite with LocalSparkContext with MockitoSug | |
| } | ||
|
|
||
| private def runTaskAndGetFailReason(taskDescription: TaskDescription): TaskFailedReason = { | ||
| runTaskGetFailReasonAndExceptionHandler(taskDescription)._1 | ||
| } | ||
|
|
||
| private def runTaskGetFailReasonAndExceptionHandler( | ||
| taskDescription: TaskDescription): (TaskFailedReason, UncaughtExceptionHandler) = { | ||
| val mockBackend = mock[ExecutorBackend] | ||
| val mockUncaughtExceptionHandler = mock[UncaughtExceptionHandler] | ||
| var executor: Executor = null | ||
| try { | ||
| executor = new Executor("id", "localhost", SparkEnv.get, userClassPath = Nil, isLocal = true, | ||
| uncaughtExceptionHandler = mockUncaughtExceptionHandler) | ||
| // the task will be launched in a dedicated worker thread | ||
| executor.launchTask(mockBackend, taskDescription) | ||
| eventually(timeout(5.seconds), interval(10.milliseconds)) { | ||
| assert(executor.numRunningTasks === 0) | ||
| } | ||
| } finally { | ||
|
|
@@ -193,7 +273,56 @@ class ExecutorSuite extends SparkFunSuite with LocalSparkContext with MockitoSug | |
| assert(statusCaptor.getAllValues().get(0).remaining() === 0) | ||
| // second update is more interesting | ||
| val failureData = statusCaptor.getAllValues.get(1) | ||
| val failReason = | ||
| SparkEnv.get.closureSerializer.newInstance().deserialize[TaskFailedReason](failureData) | ||
| (failReason, mockUncaughtExceptionHandler) | ||
| } | ||
| } | ||
|
|
||
| class FetchFailureThrowingRDD(sc: SparkContext) extends RDD[Int](sc, Nil) { | ||
| override def compute(split: Partition, context: TaskContext): Iterator[Int] = { | ||
| new Iterator[Int] { | ||
| override def hasNext: Boolean = true | ||
| override def next(): Int = { | ||
| throw new FetchFailedException( | ||
| bmAddress = BlockManagerId("1", "hostA", 1234), | ||
| shuffleId = 0, | ||
| mapId = 0, | ||
| reduceId = 0, | ||
| message = "fake fetch failure" | ||
| ) | ||
| } | ||
| } | ||
| } | ||
| override protected def getPartitions: Array[Partition] = { | ||
| Array(new SimplePartition) | ||
| } | ||
| } | ||
|
|
||
| class SimplePartition extends Partition { | ||
| override def index: Int = 0 | ||
| } | ||
|
|
||
| class FetchFailureHidingRDD( | ||
| sc: SparkContext, | ||
| val input: FetchFailureThrowingRDD, | ||
| throwOOM: Boolean) extends RDD[Int](input) { | ||
| override def compute(split: Partition, context: TaskContext): Iterator[Int] = { | ||
| val inItr = input.compute(split, context) | ||
| try { | ||
| Iterator(inItr.size) | ||
| } catch { | ||
| case t: Throwable => | ||
| if (throwOOM) { | ||
| throw new OutOfMemoryError("OOM while handling another exception") | ||
| } else { | ||
| throw new RuntimeException("User Exception that hides the original exception", t) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| override protected def getPartitions: Array[Partition] = { | ||
| Array(new SimplePartition) | ||
| } | ||
| } | ||
|
|
||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
tiny nit: but does it make sense to store the taskFailedReason (rather than the actual exception) in the task context?