Delete a file for a block if it already exists.

1 files changed, 8 insertions, 2 deletions

diff --git a/core/src/main/scala/spark/storage/DiskStore.scala b/core/src/main/scala/spark/storage/DiskStore.scala
index c7281200e7..2be5d01e31 100644
--- a/core/src/main/scala/spark/storage/DiskStore.scala
+++ b/core/src/main/scala/spark/storage/DiskStore.scala
@@ -195,9 +195,15 @@ private class DiskStore(blockManager: BlockManager, rootDirs: String)
   }
 
   private def createFile(blockId: String, allowAppendExisting: Boolean = false): File = {
-    val file = getFile(blockId)
+    var file = getFile(blockId)
     if (!allowAppendExisting && file.exists()) {
-      throw new Exception("File for block " + blockId + " already exists on disk: " + file)
+      // NOTE(shivaram): Delete the file if it exists. This might happen if a ShuffleMap task
+      // was rescheduled on the same machine as the old task ?
+      logWarning("File for block " + blockId + " already exists on disk: " + file + ". Deleting")
+      file.delete()
+      // Reopen the file
+      file = getFile(blockId)
+      // throw new Exception("File for block " + blockId + " already exists on disk: " + file)
     }
     file
   }