Fix tests

Author: Sital Kedia

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

@@ -426,16 +426,14 @@ private[spark] class MapOutputTrackerMaster(conf: SparkConf,
 
   /** Get the epoch for map output for a shuffle, if it is available */
   def getEpochForMapOutput(shuffleId: Int, mapId: Int): Option[Long] = {
-    val arrayOpt = mapStatuses.get(shuffleId)
-    if (arrayOpt.isDefined && arrayOpt.get != null) {
-      val array = arrayOpt.get
-      array.synchronized {
-        if (array(mapId) != null) {
-          return Some(epochForMapStatus(shuffleId)(mapId))
-        }
-      }
+    if (mapId < 0) {
+      return None
     }
-    None
+    for {
+      mapStatus <- mapStatuses.get(shuffleId).flatMap { mapStatusArray =>
+        Option(mapStatusArray(mapId))
+      }
+    } yield epochForMapStatus(shuffleId)(mapId)
   }
 
   /**

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

@@ -1295,7 +1295,6 @@ class DAGScheduler(
         val failedStage = stageIdToStage(task.stageId)
         val mapStage = shuffleIdToMapStage(shuffleId)
 
-        val epochForMapOutput = mapOutputTracker.getEpochForMapOutput(shuffleId, mapId)
         // It is possible that the map output was regenerated by rerun of the stage and the
         // fetch failure is being reported for stale map output. In that case, we should just
         // ignore the fetch failure and relaunch the task with latest map output info.