added an extra condition for dropping blocks in evictBlocksToFreeSpace to ensure correctness

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

@@ -478,7 +478,8 @@ private[spark] class MemoryStore(
             blockInfoManager.blockManager.stageExInfos.get(blockManager.currentStage) match {
               case Some(curStageExInfo) =>
                 // cur is this stage's output RDD
-                if (!curStageExInfo.curRunningRddMap.contains(bxi.blockId.asRDDId.get.rddId)) {
+                if (blockInfoManager.lockForWriting(bxi.blockId, blocking = false).isDefined &&
+                    !curStageExInfo.curRunningRddMap.contains(bxi.blockId.asRDDId.get.rddId)) {
                   selectedBlocks += bxi.blockId
                   freedMemory += bxi.size
                 }