fix to issue of closing channel

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

@@ -18,10 +18,11 @@
 package org.apache.spark.shuffle.sort;
 
 import java.io.File;
-import java.io.IOException;
 import java.io.FileInputStream;
+import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.channels.FileChannel;
+import java.nio.channels.WritableByteChannel;
 import javax.annotation.Nullable;
 
 import org.apache.spark.api.shuffle.ShufflePartitionWriter;
@@ -203,15 +204,14 @@ private long[] writePartitionedData(ShuffleMapOutputWriter mapOutputWriter) thro
         boolean copyThrewException = true;
         ShufflePartitionWriter writer = mapOutputWriter.getNextPartitionWriter();
         if (transferToEnabled) {
-          try (FileChannel outputChannel = writer.openChannel()) {
-            if (file.exists()) {
-              FileInputStream in = new FileInputStream(file);
-              try (FileChannel inputChannel = in.getChannel()){
-                Utils.copyFileStreamNIO(inputChannel, outputChannel, 0, inputChannel.size());
-                copyThrewException = false;
-              } finally {
-                Closeables.close(in, copyThrewException);
-              }
+          WritableByteChannel outputChannel = writer.openChannel();
+          if (file.exists()) {
+            FileInputStream in = new FileInputStream(file);
+            try (FileChannel inputChannel = in.getChannel()){
+              Utils.copyFileStreamNIO(inputChannel, outputChannel, 0, inputChannel.size());
+              copyThrewException = false;
+            } finally {
+              Closeables.close(in, copyThrewException);
             }
           }
         } else {

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -24,7 +24,7 @@ import java.lang.reflect.InvocationTargetException
 import java.math.{MathContext, RoundingMode}
 import java.net._
 import java.nio.ByteBuffer
-import java.nio.channels.{Channels, FileChannel}
+import java.nio.channels.{Channels, FileChannel, WritableByteChannel}
 import java.nio.charset.StandardCharsets
 import java.nio.file.Files
 import java.security.SecureRandom
@@ -337,10 +337,10 @@ private[spark] object Utils extends Logging {
 
   def copyFileStreamNIO(
       input: FileChannel,
-      output: FileChannel,
+      output: WritableByteChannel,
       startPosition: Long,
       bytesToCopy: Long): Unit = {
-    val initialPos = output.position()
+//  val initialPos = output.position()
     var count = 0L
     // In case transferTo method transferred less data than we have required.
     while (count < bytesToCopy) {
@@ -349,21 +349,21 @@ private[spark] object Utils extends Logging {
     assert(count == bytesToCopy,
       s"request to copy $bytesToCopy bytes, but actually copied $count bytes.")
 
-    // Check the position after transferTo loop to see if it is in the right position and
-    // give user information if not.
-    // Position will not be increased to the expected length after calling transferTo in
-    // kernel version 2.6.32, this issue can be seen in
-    // https://bugs.openjdk.java.net/browse/JDK-7052359
-    // This will lead to stream corruption issue when using sort-based shuffle (SPARK-3948).
-    val finalPos = output.position()
-    val expectedPos = initialPos + bytesToCopy
-    assert(finalPos == expectedPos,
-      s"""
-         |Current position $finalPos do not equal to expected position $expectedPos
-         |after transferTo, please check your kernel version to see if it is 2.6.32,
-         |this is a kernel bug which will lead to unexpected behavior when using transferTo.
-         |You can set spark.file.transferTo = false to disable this NIO feature.
-           """.stripMargin)
+//    // Check the position after transferTo loop to see if it is in the right position and
+//    // give user information if not.
+//    // Position will not be increased to the expected length after calling transferTo in
+//    // kernel version 2.6.32, this issue can be seen in
+//    // https://bugs.openjdk.java.net/browse/JDK-7052359
+//    // This will lead to stream corruption issue when using sort-based shuffle (SPARK-3948).
+//    val finalPos = output.position()
+//    val expectedPos = initialPos + bytesToCopy
+//    assert(finalPos == expectedPos,
+//      s"""
+//         |Current position $finalPos do not equal to expected position $expectedPos
+//         |after transferTo, please check your kernel version to see if it is 2.6.32,
+//         |this is a kernel bug which will lead to unexpected behavior when using transferTo.
+//         |You can set spark.file.transferTo = false to disable this NIO feature.
+//           """.stripMargin)
   }
 
   /**