Repository: Alluxio/alluxio (language: java)

Issue:
Accessing through the Hadoop FileContext API will cause client-side memory leaks.
**Alluxio Version:**
2.x

**Describe the bug**
We found Full GC and OOM behavior on YARN Nodemanger in an environment. after analyzing the heap dump, we found that the ConfigHashSync of the alluxio client and related classes were not being garbage collected properly.
![image](https://github.com/Alluxio/alluxio/assets/45898769/2f028c59-d6bd-4cf2-ae49-d12df4f663b8)
![image](https://github.com/Alluxio/alluxio/assets/45898769/e5b717e9-fb59-42e7-968d-d77515501837)

We believe this is due to the client not being properly closed, so we modified the `BaseFileSystem` class in the client code to see where the client was not properly closed.

```java
  public BaseFileSystem(FileSystemContext fsContext) {
    // ...
    LOG.info("constructing BaseFileSystem with context id {}", mFsContext.getId(), new Exception("print stack trace"));
  }
  public synchronized void close() throws IOException {
      // ...
    LOG.info("closing BaseFileSystem with context id {}", mFsContext.getId(), new Exception("print stack trace"));
  }
```
In the end, we get the following log:

```text
2024-01-02 11:17:19,817 [2146392] - INFO  [LogAggregationService #56:BaseFileSystem@109] - constructing BaseFileSystem with context id app-2020615593602897283
java.lang.Exception: print stack trace
	at alluxio.client.file.BaseFileSystem.<init>(BaseFileSystem.java:109)
	at alluxio.client.file.FileSystem$Factory.create(FileSystem.java:146)
	at alluxio.client.file.FileSystem$Factory.create(FileSystem.java:127)
	at alluxio.hadoop.AbstractFileSystem.initialize(AbstractFileSystem.java:533)
	at alluxio.hadoop.AbstractFileSystem.initialize(AbstractFileSystem.java:472)
	at org.apache.hadoop.fs.DelegateToFileSystem.<init>(DelegateToFileSystem.java:52)
	at alluxio.hadoop.AlluxioFileSystem.<init>(AlluxioFileSystem.java:50)
	at sun.reflect.GeneratedConstructorAccessor64.newInstance(Unknown Source)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at org.apache.hadoop.fs.AbstractFileSystem.newInstance(AbstractFileSystem.java:135)
	at org.apache.hadoop.fs.AbstractFileSystem.createFileSystem(AbstractFileSystem.java:173)
	at org.apache.hadoop.fs.AbstractFileSystem.get(AbstractFileSystem.java:258)
	at org.apache.hadoop.fs.FileContext$2.run(FileContext.java:342)
	at org.apache.hadoop.fs.FileContext$2.run(FileContext.java:339)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1893)
	at org.apache.hadoop.fs.FileContext.getAbstractFileSystem(FileContext.java:339)
	at org.apache.hadoop.fs.FileContext.getFileContext(FileContext.java:465)
	at org.apache.hadoop.fs.FileContext.getFileContext(FileContext.java:491)
	at org.apache.hadoop.fs.FileContext.getFileContext(FileContext.java:461)
	at org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat$LogWriter$1.run(AggregatedLogFormat.java:476)
	at org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat$LogWriter$1.run(AggregatedLogFormat.java:473)
	at java.security.AccessController.doPrivileged(Native Method)
	at javax.security.auth.Subject.doAs(Subject.java:422)
	at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1893)
	at org.apache.hadoop.yarn.logaggregation.AggregatedLogFormat$LogWriter.initialize(AggregatedLogFormat.java:472)
	at org.apache.hadoop.yarn.logaggregation.filecontroller.tfile.LogAggregationTFileController.initializeWriter(LogAggregationTFileController.java:90)
	at org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AppLogAggregatorImpl.uploadLogsForContainers(AppLogAggregatorImpl.java:306)
	at org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AppLogAggregatorImpl.doAppLogAggregation(AppLogAggregatorImpl.java:459)
	at org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.AppLogAggregatorImpl.run(AppLogAggregatorImpl.java:415)
	at org.apache.hadoop.yarn.server.nodemanager.containermanager.logaggregation.LogAggregationService$1.run(LogAggregationService.java:265)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

```
When the method `FileContext.getFileContext` is called, it creates an instance of [alluxio.hadoop.AlluxioFileSystem](https://github.com/Alluxio/alluxio/blob/main/dora/core/client/hdfs/src/main/java/alluxio/hadoop/AlluxioFileSystem.java) to access Alluxio. However, the base class [org.apache.hadoop.fs.AbstractFileSystem](https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/AbstractFileSystem.java) does not have a close method to release the held resources, which can lead to memory leaks.

**To Reproduce**
```java
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;

import java.net.URI;

public class Main {
    public static void main(String[] args) throws Exception {
        while (true){
            Configuration conf = new Configuration();
            conf.set("fs.AbstractFileSystem.alluxio.impl", "alluxio.hadoop.AlluxioFileSystem");
            // every call will create new filesystem but not close
            FileContext fileContext = FileContext.getFileContext(URI.create("alluxio://localhost:19998/"), conf);
            FileStatus fileStatus = fileContext.getFileStatus(new Path("/"));
            System.out.println(fileStatus);
            Thread.sleep(200);
        }
    }
}
```

**Expected behavior**
Will not produce memory leaks.

**Urgency**
High.

**Are you planning to fix it**
yes.

**Additional context**




Discussion / hints:

Fix patch:
diff --git a/core/client/hdfs/src/main/java/alluxio/hadoop/AbstractFileSystem.java b/core/client/hdfs/src/main/java/alluxio/hadoop/AbstractFileSystem.java
index 4589dd252673..1450833ac88b 100644
--- a/core/client/hdfs/src/main/java/alluxio/hadoop/AbstractFileSystem.java
+++ b/core/client/hdfs/src/main/java/alluxio/hadoop/AbstractFileSystem.java
@@ -90,6 +90,7 @@ public abstract class AbstractFileSystem extends org.apache.hadoop.fs.FileSystem
   private Statistics mStatistics = null;
   private String mAlluxioHeader = null;
   private boolean mExcludeMountInfoOnListStatus;
+  private volatile boolean mIsClosed = false;
 
   /**
    * Constructs a new {@link AbstractFileSystem} instance with specified a {@link FileSystem}
@@ -143,12 +144,17 @@ public FSDataOutputStream append(Path path, int bufferSize, Progressable progres
 
   @Override
   public void close() throws IOException {
+    if (mIsClosed) {
+      return;
+    }
+
     // super.close should be called first before releasing the resources in this instance, as the
     // super class may invoke other methods in this class. For example,
     // org.apache.hadoop.fs.FileSystem.close may check the existence of certain temp files before
     // closing
     super.close();
     mFileSystem.close();
+    mIsClosed = true;
   }
 
   /**
diff --git a/core/client/hdfs/src/main/java/alluxio/hadoop/AlluxioFileSystem.java b/core/client/hdfs/src/main/java/alluxio/hadoop/AlluxioFileSystem.java
index afa810056b44..d5945b9c557c 100644
--- a/core/client/hdfs/src/main/java/alluxio/hadoop/AlluxioFileSystem.java
+++ b/core/client/hdfs/src/main/java/alluxio/hadoop/AlluxioFileSystem.java
@@ -48,4 +48,12 @@ public class AlluxioFileSystem extends DelegateToFileSystem {
       throws IOException, URISyntaxException {
     super(uri, new FileSystem(), conf, Constants.SCHEME, false);
   }
+
+  // CHECKSTYLE.OFF: NoFinalizer
+  @Override
+  protected void finalize() throws Throwable {
+    fsImpl.close();
+    super.finalize();
+  }
+  // CHECKSTYLE.ON: NoFinalizer
 }


Test patch: