Merge pull request #618 from palantir/yh/removal-check-hack

[SPARK-25299] Check for existence outside executor before removing
palantir · Oct 31, 2019 · af0ad25 · af0ad25
2 parents 45b0d28 + ed95788
commit af0ad25
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -463,8 +463,14 @@ private[spark] class MapOutputTrackerMaster(
   def unregisterMapOutput(shuffleId: Int, mapId: Int, bmAddress: BlockManagerId) {
     shuffleStatuses.get(shuffleId) match {
       case Some(shuffleStatus) =>
-        shuffleStatus.removeMapOutput(mapId, bmAddress)
-        incrementEpoch()
+        if (shuffleStatus.mapStatuses(mapId) != null &&
+            !shuffleDriverComponents.checkIfMapOutputStoredOutsideExecutor(
+              shuffleId,
+              mapId,
+              shuffleStatus.mapStatuses(mapId).mapTaskAttemptId)) {
+          shuffleStatus.removeMapOutput(mapId, bmAddress)
+          incrementEpoch()
+        }
       case None =>
         throw new SparkException("unregisterMapOutput called for nonexistent shuffle ID")
     }