MetadataFetchFailedException extends FetchFailedException.

Author: rxin

core/src/main/scala/org/apache/spark/TaskEndReason.scala

@@ -38,7 +38,7 @@ case object Resubmitted extends TaskEndReason // Task was finished earlier but w
 
 @DeveloperApi
 case class FetchFailed(
-    bmAddress: BlockManagerId,
+    bmAddress: BlockManagerId,  // Note that bmAddress can be null
     shuffleId: Int,
     mapId: Int,
     reduceId: Int)

core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala

@@ -21,6 +21,10 @@ import java.nio.ByteBuffer
 
 import org.apache.spark.util.SerializableBuffer
 
+/**
+ * Description of a task that gets passed onto executors to be executed, usually created by
+ * [[TaskSetManager.resourceOffer]].
+ */
 private[spark] class TaskDescription(
     val taskId: Long,
     val executorId: String,

core/src/main/scala/org/apache/spark/shuffle/FetchFailedException.scala

@@ -23,6 +23,8 @@ import org.apache.spark.{FetchFailed, TaskEndReason}
 /**
  * Failed to fetch a shuffle block. The executor catches this exception and propagates it
  * back to DAGScheduler (through TaskEndReason) so we'd resubmit the previous stage.
+ *
+ * Note that bmAddress can be null.
  */
 private[spark] class FetchFailedException(
     bmAddress: BlockManagerId,
@@ -36,3 +38,15 @@ private[spark] class FetchFailedException(
 
   def toTaskEndReason: TaskEndReason = FetchFailed(bmAddress, shuffleId, mapId, reduceId)
 }
+
+/**
+ * Failed to get shuffle metadata from [[org.apache.spark.MapOutputTracker]].
+ */
+private[spark] class MetadataFetchFailedException(
+    shuffleId: Int,
+    reduceId: Int,
+    message: String)
+  extends FetchFailedException(null, shuffleId, -1, reduceId) {
+
+  override def getMessage: String = message
+}