May be fix tests?

apache · Mar 28, 2017 · 1e6e88a · 1e6e88a
1 parent d32856d
commit 1e6e88a
Showing 1 changed file with 9 additions and 4 deletions.
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -426,11 +426,16 @@ 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] = {
-    for {
-      mapStatus <- mapStatuses.get(shuffleId).flatMap { mapStatusArray =>
-        Option(mapStatusArray(mapId))
+    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))
+        }
       }
-    } yield epochForMapStatus(shuffleId)(mapId)
+    }
+    None
   }
 
   /**