rather than requiring all tmp files to exist, just write a zero-lengt…

…h dest file either way

core/src/main/scala/org/apache/spark/shuffle/ShuffleOutputCoordinator.scala

@@ -37,13 +37,16 @@ private[spark] object ShuffleOutputCoordinator extends Logging {
    * destinations, and return (true, the given MapStatus).  If all destination files exist, then
    * delete all temporary files, and return (false, the MapStatus from previously committed shuffle
    * output).
-
-   * @param shuffleId
-   * @param partitionId
+   *
+   * Note that this will write to all destination files.  If the tmp file is missing, then a
+   * zero-length destination file will be created.  This is so the ShuffleOutputCoordinator can work
+   * even when there is a non-determinstic data, where the output exists in one attempt, but is
+   * empty in another attempt.
+   *
    * @param tmpToDest  Seq of (temporary, destination) file pairs
    * @param mapStatus the [[MapStatus]] for the output already written to the the temporary files
-   * @return pair of (true iff the set of temporary files was moved to the destination, the
-   *         MapStatus of the winn
+   * @return pair of: (1) true iff the set of temporary files was moved to the destination and (2)
+   *         the MapStatus of the committed attempt.
    *
    */
   def commitOutputs(
@@ -66,8 +69,6 @@ private[spark] object ShuffleOutputCoordinator extends Logging {
       mapStatus: MapStatus,
       mapStatusFile: File,
       serializer: SerializerInstance): (Boolean, MapStatus) = synchronized {
-    tmpToDest.foreach { case (tmp, _) => require(tmp.exists(), s"Cannot commit non-existent " +
-      s"shuffle output $tmp -- must be at least a zero-length file.")}
     val destAlreadyExists = tmpToDest.forall{_._2.exists()} && mapStatusFile.exists()
     if (!destAlreadyExists) {
       tmpToDest.foreach { case (tmp, dest) =>
@@ -78,7 +79,14 @@ private[spark] object ShuffleOutputCoordinator extends Logging {
         if (dest.exists()) {
           dest.delete()
         }
-        tmp.renameTo(dest)
+        if (tmp.exists()) {
+          tmp.renameTo(dest)
+        } else {
+          // we always create the destination files, so this works correctly even when the
+          // input data is non-deterministic (potentially empty in one iteration, and non-empty
+          // in another)
+          dest.createNewFile()
+        }
       }
       val out = serializer.serializeStream(new FileOutputStream(mapStatusFile))
       out.writeObject(mapStatus)

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

@@ -133,11 +133,6 @@ private[spark] class DiskBlockObjectWriter(
       // In certain compression codecs, more bytes are written after close() is called
       writeMetrics.incShuffleBytesWritten(finalPosition - reportedPosition)
     } else {
-      if (!file.exists()) {
-        // SPARK-8029 -- we need to write a zero-length file so we can commit the same set of files
-        // on all attempts (even if the data is non-deterministic)
-        file.createNewFile()
-      }
       finalPosition = file.length()
     }
     commitAndCloseHasBeenCalled = true

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

@@ -679,12 +679,6 @@ private[spark] class ExternalSorter[K, V, C](
       }
     }
 
-    // SPARK-8029 the ShuffleOutputCoordinator requires all shuffle output files to always exist,
-    // even if they are zero-length
-    if (!outputFile.exists()) {
-      outputFile.createNewFile()
-    }
-
     context.taskMetrics().incMemoryBytesSpilled(memoryBytesSpilled)
     context.taskMetrics().incDiskBytesSpilled(diskBytesSpilled)
     context.internalMetricsToAccumulators(

core/src/test/scala/org/apache/spark/shuffle/ShuffleOutputCoordinatorSuite.scala

@@ -31,7 +31,7 @@ class ShuffleOutputCoordinatorSuite extends SparkFunSuite with BeforeAndAfterEac
   var tempDir: File = _
   var mapStatusFile: File = _
   // use the "port" as a way to distinguish mapstatuses, just for the test
-  def mapStatus(id: Int) = MapStatus(BlockManagerId("1", "a.b.c", id), Array(0L, 1L))
+  def mapStatus(id: Int): MapStatus = MapStatus(BlockManagerId("1", "a.b.c", id), Array(0L, 1L))
   def ser: SerializerInstance = new JavaSerializer(new SparkConf()).newInstance()
 
   override def beforeEach(): Unit = {
@@ -114,11 +114,23 @@ class ShuffleOutputCoordinatorSuite extends SparkFunSuite with BeforeAndAfterEac
     secondAttempt.foreach{ case (t, d) => assert(!t.exists())}
   }
 
-  test("no missing tmp files") {
+  test("missing tmp files become zero-length destination files") {
+    val extraDestFile = new File(tempDir, "blah")
     val firstAttempt = generateAttempt(0) ++
-      Seq(new File(tempDir, "bogus") -> new File(tempDir, "blah"))
-    val ex = intercept[IllegalArgumentException] {commit(firstAttempt, 1)}
-    assert(ex.getMessage.contains("Cannot commit non-existent shuffle output"))
+      Seq(new File(tempDir, "bogus") -> extraDestFile)
+    assert(commit(firstAttempt, 1)._1)
+    verifyFiles(0)
+    assert(extraDestFile.exists())
+    assert(extraDestFile.length() === 0)
+
+    // if we attempt the move again and *only* the missing tmp file is missing, we still
+    // do the move
+    extraDestFile.delete()
+    val secondAttempt = generateAttempt(1) ++
+      Seq(new File(tempDir, "flippy") -> extraDestFile)
+    assert(commit(secondAttempt, 2)._1)
+    verifyFiles(1)
+    assert(extraDestFile.exists())
+    assert(extraDestFile.length() === 0)
   }
-
 }