From 30d95703ddd66a96dc2ddaae9ca38900583544fe Mon Sep 17 00:00:00 2001
From: yangjie01 <yangjie01@baidu.com>
Date: Sat, 14 Oct 2023 20:01:20 +0800
Subject: [PATCH] [SPARK-45530][CORE] Use `java.lang.ref.Cleaner` instead of
 `finalize` for `NioBufferedFileInputStream`

### What changes were proposed in this pull request?
This pr change to use `java.lang.ref.Cleaner` instead of `finalize()` for `NioBufferedFileInputStream`. They are all protective measures for resource cleanup, but the `finalize()` method has been marked as `deprecated` since Java 9 and will be removed in the future, `java.lang.ref.Cleaner` is the more recommended solution.

https://github.com/openjdk/jdk/blob/dfacda488bfbe2e11e8d607a6d08527710286982/src/java.base/share/classes/java/lang/Object.java#L546-L568

```
     * deprecated The finalization mechanism is inherently problematic.
     * Finalization can lead to performance issues, deadlocks, and hangs.
     * Errors in finalizers can lead to resource leaks; there is no way to cancel
     * finalization if it is no longer necessary; and no ordering is specified
     * among calls to {code finalize} methods of different objects.
     * Furthermore, there are no guarantees regarding the timing of finalization.
     * The {code finalize} method might be called on a finalizable object
     * only after an indefinite delay, if at all.
     *
     * Classes whose instances hold non-heap resources should provide a method
     * to enable explicit release of those resources, and they should also
     * implement {link AutoCloseable} if appropriate.
     * The {link java.lang.ref.Cleaner} and {link java.lang.ref.PhantomReference}
     * provide more flexible and efficient ways to release resources when an object
     * becomes unreachable.
     *
     * throws Throwable the {code Exception} raised by this method
     * see java.lang.ref.WeakReference
     * see java.lang.ref.PhantomReference
     * jls 12.6 Finalization of Class Instances
     */
    Deprecated(since="9")
    protected void finalize() throws Throwable { }
```

### Why are the changes needed?
Clean up deprecated api usage

### Does this PR introduce _any_ user-facing change?
No

### How was this patch tested?
Pass GitHub Actions

### Was this patch authored or co-authored using generative AI tooling?
No

Closes #43333 from LuciferYang/nio-buffered-fis-cleaner.

Lead-authored-by: yangjie01 <yangjie01@baidu.com>
Co-authored-by: YangJie <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>
---
 .../spark/io/NioBufferedFileInputStream.java  | 34 +++++++++++++++----
 1 file changed, 28 insertions(+), 6 deletions(-)

diff --git a/core/src/main/java/org/apache/spark/io/NioBufferedFileInputStream.java b/core/src/main/java/org/apache/spark/io/NioBufferedFileInputStream.java
index 91910b99ac999..441587cf7350e 100644
--- a/core/src/main/java/org/apache/spark/io/NioBufferedFileInputStream.java
+++ b/core/src/main/java/org/apache/spark/io/NioBufferedFileInputStream.java
@@ -18,6 +18,8 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.lang.ref.Cleaner;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.nio.file.StandardOpenOption;
@@ -32,8 +34,11 @@
  */
 public final class NioBufferedFileInputStream extends InputStream {
 
+  private static final Cleaner CLEANER = Cleaner.create();
   private static final int DEFAULT_BUFFER_SIZE_BYTES = 8192;
 
+  private final Cleaner.Cleanable cleanable;
+
   private final ByteBuffer byteBuffer;
 
   private final FileChannel fileChannel;
@@ -42,6 +47,7 @@ public NioBufferedFileInputStream(File file, int bufferSizeInBytes) throws IOExc
     byteBuffer = ByteBuffer.allocateDirect(bufferSizeInBytes);
     fileChannel = FileChannel.open(file.toPath(), StandardOpenOption.READ);
     byteBuffer.flip();
+    this.cleanable = CLEANER.register(this, new ResourceCleaner(fileChannel, byteBuffer));
   }
 
   public NioBufferedFileInputStream(File file) throws IOException {
@@ -125,13 +131,29 @@ private long skipFromFileChannel(long n) throws IOException {
 
   @Override
   public synchronized void close() throws IOException {
-    fileChannel.close();
-    StorageUtils.dispose(byteBuffer);
+    try {
+      this.cleanable.clean();
+    } catch (UncheckedIOException re) {
+      if (re.getCause() != null) {
+        throw re.getCause();
+      } else {
+        throw re;
+      }
+    }
   }
 
-  @SuppressWarnings("deprecation")
-  @Override
-  protected void finalize() throws IOException {
-    close();
+  private record ResourceCleaner(
+      FileChannel fileChannel,
+      ByteBuffer byteBuffer) implements Runnable {
+    @Override
+    public void run() {
+      try {
+        fileChannel.close();
+      } catch (IOException e) {
+        throw new UncheckedIOException(e);
+      } finally {
+        StorageUtils.dispose(byteBuffer);
+      }
+    }
   }
 }