[SPARK-21527][CORE] Use buffer limit in order to use JAVA NIO Util's buffercache

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -321,6 +321,13 @@ package object config {
       .intConf
       .createWithDefault(3)
 
+  private[spark] val BUFFER_WRITE_CHUNK_SIZE =
+    ConfigBuilder("spark.buffer.write.chunkSize")
+      .internal()
+      .doc("The block size limit when use ChunkedByteBuffer to writeFully bytes.")
+      .bytesConf(ByteUnit.BYTE)
+      .createWithDefault(64 * 1024 * 1024)
+
   private[spark] val REDUCER_MAX_REQ_SIZE_SHUFFLE_TO_MEM =
     ConfigBuilder("spark.reducer.maxReqSizeShuffleToMem")
       .internal()

core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala

@@ -24,6 +24,7 @@ import java.nio.channels.WritableByteChannel
 import com.google.common.primitives.UnsignedBytes
 import io.netty.buffer.{ByteBuf, Unpooled}
 
+import org.apache.spark.internal.config
 import org.apache.spark.network.util.ByteArrayWritableChannel
 import org.apache.spark.storage.StorageUtils
 
@@ -40,6 +41,9 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
   require(chunks != null, "chunks must not be null")
   require(chunks.forall(_.position() == 0), "chunks' positions must be 0")
 
+  // Chunk size in bytes
+  private val NIO_BUFFER_LIMIT = SparkEnv.get.conf.get(config.BUFFER_WRITE_CHUNK_SIZE)
+
   private[this] var disposed: Boolean = false
 
   /**
@@ -56,7 +60,10 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
    */
   def writeFully(channel: WritableByteChannel): Unit = {
     for (bytes <- getChunks()) {
-      while (bytes.remaining > 0) {
+      val capacity = bytes.limit()
+      while (bytes.remaining() > 0) {
+        val ioSize = Math.min(bytes.remaining(), NIO_BUFFER_LIMIT)
+        bytes.limit(bytes.position + ioSize)
         channel.write(bytes)
       }
     }