Fix NullPointerException caused by unregistered map outputs.

core/src/main/scala/spark/MapOutputTracker.scala

@@ -156,6 +156,10 @@ private[spark] class MapOutputTracker(actorSystem: ActorSystem, isMaster: Boolea
         fetching -= shuffleId
         fetching.notifyAll()
       }
+      if (fetchedStatuses.contains(null)) {
+        throw new FetchFailedException(null, shuffleId, -1, reduceId,
+          new Exception("Missing an output location for shuffle " + shuffleId))
+      }
       return fetchedStatuses.map(s =>
         (s.address, MapOutputTracker.decompressSize(s.compressedSizes(reduceId))))
     } else {

core/src/main/scala/spark/scheduler/DAGScheduler.scala

@@ -479,8 +479,10 @@ class DAGScheduler(taskSched: TaskScheduler) extends TaskSchedulerListener with
           ") for resubmision due to a fetch failure")
         // Mark the map whose fetch failed as broken in the map stage
         val mapStage = shuffleToMapStage(shuffleId)
-        mapStage.removeOutputLoc(mapId, bmAddress)
-        mapOutputTracker.unregisterMapOutput(shuffleId, mapId, bmAddress)
+        if (mapId != -1) {
+          mapStage.removeOutputLoc(mapId, bmAddress)
+          mapOutputTracker.unregisterMapOutput(shuffleId, mapId, bmAddress)
+        }
         logInfo("The failed fetch was from " + mapStage + " (" + mapStage.origin +
           "); marking it for resubmission")
         failed += mapStage