Update as commented

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

@@ -293,6 +293,13 @@ package object config {
       .booleanConf
       .createWithDefault(false)
 
+  private[spark] val BUFFER_WRITE_CHUNK_SIZE =
+    ConfigBuilder("spark.buffer.write.chunkSize")
+      .internal()
+      .doc("The chunk size during writing out the bytes of ChunkedByteBuffer.")
+      .bytesConf(ByteUnit.BYTE)
+      .createWithDefault(64 * 1024 * 1024)
+
   private[spark] val CHECKPOINT_COMPRESS =
     ConfigBuilder("spark.checkpoint.compress")
       .doc("Whether to compress RDD checkpoints. Generally a good idea. Compression will use " +
@@ -321,13 +328,6 @@ 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

@@ -42,7 +42,9 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
   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 val bufferWriteChunkSize =
+    Option(SparkEnv.get).map(_.conf.get(BUFFER_WRITE_CHUNK_SIZE))
+      .getOrElse(BUFFER_WRITE_CHUNK_SIZE.defaultValue.get)
 
   private[this] var disposed: Boolean = false
 
@@ -60,9 +62,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
    */
   def writeFully(channel: WritableByteChannel): Unit = {
     for (bytes <- getChunks()) {
-      val capacity = bytes.limit()
       while (bytes.remaining() > 0) {
-        val ioSize = Math.min(bytes.remaining(), NIO_BUFFER_LIMIT)
+        val ioSize = Math.min(bytes.remaining(), bufferWriteChunkSize)
         bytes.limit(bytes.position + ioSize)
         channel.write(bytes)
       }