add a new method to avoid file truncate

Author: LuciferYang

core/src/main/java/org/apache/spark/shuffle/sort/BypassMergeSortShuffleWriter.java

@@ -289,10 +289,7 @@ public Option<MapStatus> stop(boolean success) {
           try {
             for (DiskBlockObjectWriter writer : partitionWriters) {
               // This method explicitly does _not_ throw exceptions:
-              File file = writer.revertPartialWritesAndClose();
-              if (!file.delete()) {
-                logger.error("Error while deleting file {}", file.getAbsolutePath());
-              }
+              writer.deleteHeldFile();
             }
           } finally {
             partitionWriters = null;

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

@@ -262,6 +262,27 @@ private[spark] class DiskBlockObjectWriter(
     file
   }
 
+  /**
+   * Reverts write metrics that haven't been committed yet and delete the file held
+   * by current `DiskBlockObjectWriter`. Callers should invoke this function when there
+   * are runtime exceptions in file writing process and the file is no longer needed.
+   */
+  def deleteHeldFile(): Unit = {
+    Utils.tryWithSafeFinally {
+      if (initialized) {
+        writeMetrics.decBytesWritten(reportedPosition - committedPosition)
+        writeMetrics.decRecordsWritten(numRecordsWritten)
+        closeResources()
+      }
+    } {
+      if (file.exists()) {
+        if (!file.delete()) {
+          logWarning(s"Error deleting $file")
+        }
+      }
+    }
+  }
+
   /**
    * Writes a key-value pair.
    */

core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala

@@ -250,12 +250,7 @@ class ExternalAppendOnlyMap[K, V, C](
       if (!success) {
         // This code path only happens if an exception was thrown above before we set success;
         // close our stuff and let the exception be thrown further
-        writer.revertPartialWritesAndClose()
-        if (file.exists()) {
-          if (!file.delete()) {
-            logWarning(s"Error deleting ${file}")
-          }
-        }
+        writer.deleteHeldFile()
       }
     }
 

core/src/main/scala/org/apache/spark/util/collection/ExternalSorter.scala

@@ -321,12 +321,7 @@ private[spark] class ExternalSorter[K, V, C](
       if (!success) {
         // This code path only happens if an exception was thrown above before we set success;
         // close our stuff and let the exception be thrown further
-        writer.revertPartialWritesAndClose()
-        if (file.exists()) {
-          if (!file.delete()) {
-            logWarning(s"Error deleting ${file}")
-          }
-        }
+        writer.deleteHeldFile()
       }
     }