Avoid writing empty files in BypassMergeSortShuffleWriter

Author: JoshRosen

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

@@ -108,7 +108,7 @@ public void insertAll(Iterator<Product2<K, V>> records) throws IOException {
       final File file = tempShuffleBlockIdPlusFile._2();
       final BlockId blockId = tempShuffleBlockIdPlusFile._1();
       partitionWriters[i] =
-        blockManager.getDiskWriter(blockId, file, serInstance, fileBufferSize, writeMetrics).open();
+        blockManager.getDiskWriter(blockId, file, serInstance, fileBufferSize, writeMetrics);
     }
     // Creating the file to write to and creating a disk writer both involve interacting with
     // the disk, and can take a long time in aggregate when we open many files, so should be
@@ -143,6 +143,9 @@ public long[] writePartitionedFile(
     boolean threwException = true;
     try {
       for (int i = 0; i < numPartitions; i++) {
+        if (partitionWriters[i].fileSegment().length() == 0) {
+          continue;
+        }
         final FileInputStream in = new FileInputStream(partitionWriters[i].fileSegment().file());
         boolean copyThrewException = true;
         try {