Check for existing executor

Author: Kimahriman

core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala

@@ -317,14 +317,17 @@ class BlockManagerMasterEndpoint(
     val blocksToDeleteByShuffleService =
       new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
     mapOutputTracker.shuffleStatuses.get(shuffleId).map { shuffleStatus =>
-      shuffleStatus.mapStatuses
-        .filter(_.location.port == externalShuffleServicePort)
-        .foreach { mapStatus =>
+      shuffleStatus.mapStatuses.foreach { mapStatus =>
+        // Port should always be external shuffle port if external shuffle is enabled
+        val isShufflePort = mapStatus.location.port == externalShuffleServicePort
+        val executorDeallocated = !blockManagerIdByExecutor.contains(mapStatus.location.executorId)
+        if (isShufflePort && executorDeallocated) {
           val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
             new mutable.HashSet[BlockId])
           val blocksToDel =
             shuffleManager.shuffleBlockResolver.getBlocksForShuffle(shuffleId, mapStatus.mapId)
           blocksToDel.foreach(blocks.add(_))
+        }
       }
     }