[SPARK-49509][CORE] Use Platform.allocateDirectBuffer instead of ByteBuffer.allocateDirect

This PR aims to use `Platform.allocateDirectBuffer` instead of `ByteBuffer.allocateDirect`.

#47733 (review)

Allocating off-heap memory should use the `allocateDirectBuffer` API provided `by Platform`.

No

GA

No

Closes #47987 from cxzl25/SPARK-49509.

Authored-by: sychen <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit 2ed6c3e)
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: cxzl25

core/src/main/java/org/apache/spark/io/NioBufferedFileInputStream.java

@@ -14,6 +14,7 @@
 package org.apache.spark.io;
 
 import org.apache.spark.storage.StorageUtils;
+import org.apache.spark.unsafe.Platform;
 
 import java.io.File;
 import java.io.IOException;
@@ -39,7 +40,7 @@ public final class NioBufferedFileInputStream extends InputStream {
   private final FileChannel fileChannel;
 
   public NioBufferedFileInputStream(File file, int bufferSizeInBytes) throws IOException {
-    byteBuffer = ByteBuffer.allocateDirect(bufferSizeInBytes);
+    byteBuffer = Platform.allocateDirectBuffer(bufferSizeInBytes);
     fileChannel = FileChannel.open(file.toPath(), StandardOpenOption.READ);
     byteBuffer.flip();
   }

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

@@ -34,6 +34,7 @@ import org.apache.spark.internal.{config, Logging}
 import org.apache.spark.network.buffer.ManagedBuffer
 import org.apache.spark.network.util.{AbstractFileRegion, JavaUtils}
 import org.apache.spark.security.CryptoStreamUtils
+import org.apache.spark.unsafe.Platform
 import org.apache.spark.unsafe.array.ByteArrayMethods
 import org.apache.spark.util.Utils
 import org.apache.spark.util.io.ChunkedByteBuffer
@@ -309,7 +310,7 @@ private class ReadableChannelFileRegion(source: ReadableByteChannel, blockSize:
 
   private var _transferred = 0L
 
-  private val buffer = ByteBuffer.allocateDirect(64 * 1024)
+  private val buffer = Platform.allocateDirectBuffer(64 * 1024)
   buffer.flip()
 
   override def count(): Long = blockSize