[Fix](hdfs-fs)The cache expiration should explicitly release the held fs

fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFSPhantomManager.java

@@ -0,0 +1,117 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.fs.remote;
+
+import org.apache.doris.common.CustomThreadFactory;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.IOException;
+import java.lang.ref.PhantomReference;
+import java.lang.ref.Reference;
+import java.lang.ref.ReferenceQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * The RemoteFSPhantomManager class is responsible for managing the phantom references
+ * of RemoteFileSystem objects. It ensures that the associated FileSystem resources are
+ * automatically cleaned up when the RemoteFileSystem objects are garbage collected.
+ * <p>
+ * By utilizing a ReferenceQueue and PhantomReference, this class can monitor the lifecycle
+ * of RemoteFileSystem objects. When a RemoteFileSystem object is no longer in use and is
+ * garbage collected, its corresponding FileSystem resource is properly closed to prevent
+ * resource leaks.
+ * <p>
+ * The class provides a thread-safe mechanism to ensure that the cleanup thread is started only once.
+ * <p>
+ * Main functionalities include:
+ * - Registering phantom references of RemoteFileSystem objects.
+ * - Starting a periodic cleanup thread that automatically closes unused FileSystem resources.
+ */
+public class RemoteFSPhantomManager {
+
+    private static final Logger LOG = LogManager.getLogger(RemoteFSPhantomManager.class);
+
+    // Scheduled executor for periodic resource cleanup
+    private static ScheduledExecutorService cleanupExecutor;
+
+    // Reference queue for monitoring RemoteFileSystem objects' phantom references
+    private static final ReferenceQueue<RemoteFileSystem> referenceQueue = new ReferenceQueue<>();
+
+    // Map storing the phantom references and their corresponding FileSystem objects
+    private static final ConcurrentHashMap<PhantomReference<RemoteFileSystem>, FileSystem> referenceMap
+            = new ConcurrentHashMap<>();
+
+    // Flag indicating whether the cleanup thread has been started
+    private static final AtomicBoolean isStarted = new AtomicBoolean(false);
+
+    /**
+     * Registers a phantom reference for a RemoteFileSystem object in the manager.
+     * If the cleanup thread has not been started, it will be started.
+     *
+     * @param remoteFileSystem the RemoteFileSystem object to be registered
+     */
+    public static void registerPhantomReference(RemoteFileSystem remoteFileSystem) {
+        if (!isStarted.get()) {
+            start();
+            isStarted.set(true);
+        }
+        RemoteFileSystemPhantomReference phantomReference = new RemoteFileSystemPhantomReference(remoteFileSystem,
+                referenceQueue);
+        referenceMap.put(phantomReference, remoteFileSystem.dfsFileSystem);
+    }
+
+    /**
+     * Starts the cleanup thread, which periodically checks and cleans up unused FileSystem resources.
+     * The method uses double-checked locking to ensure thread-safe startup of the cleanup thread.
+     */
+    public static void start() {
+        if (isStarted.compareAndSet(false, true)) {
+            synchronized (RemoteFSPhantomManager.class) {
+                LOG.info("Starting cleanup thread for RemoteFileSystem objects");
+                if (cleanupExecutor == null) {
+                    CustomThreadFactory threadFactory = new CustomThreadFactory("remote-fs-phantom-cleanup");
+                    cleanupExecutor = Executors.newScheduledThreadPool(1, threadFactory);
+                    cleanupExecutor.scheduleAtFixedRate(() -> {
+                        Reference<? extends RemoteFileSystem> ref;
+                        while ((ref = referenceQueue.poll()) != null) {
+                            RemoteFileSystemPhantomReference phantomRef = (RemoteFileSystemPhantomReference) ref;
+
+                            FileSystem fs = referenceMap.remove(phantomRef);
+                            if (fs != null) {
+                                try {
+                                    fs.close();
+                                    LOG.info("Closed file system: {}", fs.getUri());
+                                } catch (IOException e) {
+                                    LOG.warn("Failed to close file system", e);
+                                }
+                            }
+                        }
+                    }, 0, 1, TimeUnit.MINUTES);
+                }
+            }
+        }
+    }
+
+}

fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystem.java

@@ -30,15 +30,20 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
 
+import java.io.Closeable;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantLock;
 
-public abstract class RemoteFileSystem extends PersistentFileSystem {
+public abstract class RemoteFileSystem extends PersistentFileSystem implements Closeable {
     // this field will be visited by multi-threads, better use volatile qualifier
     protected volatile org.apache.hadoop.fs.FileSystem dfsFileSystem = null;
+    private final ReentrantLock fsLock = new ReentrantLock();
+    protected static final AtomicBoolean closed = new AtomicBoolean(false);
 
     public RemoteFileSystem(String name, StorageBackend.StorageType type) {
         super(name, type);
@@ -120,4 +125,18 @@ public Status renameDir(String origFilePath,
 
         return rename(origFilePath, destFilePath);
     }
+
+    @Override
+    public void close() throws IOException {
+        fsLock.lock();
+        try {
+            if (!closed.getAndSet(true)) {
+                if (dfsFileSystem != null) {
+                    dfsFileSystem.close();
+                }
+            }
+        } finally {
+            fsLock.unlock();
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/fs/remote/RemoteFileSystemPhantomReference.java

@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.fs.remote;
+
+import org.apache.hadoop.fs.FileSystem;
+
+import java.lang.ref.PhantomReference;
+import java.lang.ref.ReferenceQueue;
+
+public class RemoteFileSystemPhantomReference extends PhantomReference<RemoteFileSystem> {
+
+    private FileSystem fs;
+
+    /**
+     * Creates a new phantom reference that refers to the given object and
+     * is registered with the given queue.
+     *
+     * <p> It is possible to create a phantom reference with a {@code null}
+     * queue.  Such a reference will never be enqueued.
+     *
+     * @param referent the object the new phantom reference will refer to
+     * @param q        the queue with which the reference is to be registered,
+     *                 or {@code null} if registration is not required
+     */
+    public RemoteFileSystemPhantomReference(RemoteFileSystem referent, ReferenceQueue<? super RemoteFileSystem> q) {
+        super(referent, q);
+        this.fs = referent.dfsFileSystem;
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java

@@ -58,8 +58,15 @@ private void initFsProperties() {
 
     @Override
     protected FileSystem nativeFileSystem(String remotePath) throws UserException {
+        //todo Extracting a common method to achieve logic reuse
+        if (closed.get()) {
+            throw new UserException("FileSystem is closed.");
+        }
         if (dfsFileSystem == null) {
             synchronized (this) {
+                if (closed.get()) {
+                    throw new UserException("FileSystem is closed.");
+                }
                 if (dfsFileSystem == null) {
                     Configuration conf = DFSFileSystem.getHdfsConf(ifNotSetFallbackToSimpleAuth());
                     System.setProperty("com.amazonaws.services.s3.enableV4", "true");
@@ -72,6 +79,7 @@ protected FileSystem nativeFileSystem(String remotePath) throws UserException {
                     } catch (Exception e) {
                         throw new UserException("Failed to get S3 FileSystem for " + e.getMessage(), e);
                     }
+                    RemoteFSPhantomManager.registerPhantomReference(this);
                 }
             }
         }

fe/fe-core/src/main/java/org/apache/doris/fs/remote/dfs/DFSFileSystem.java

@@ -26,6 +26,7 @@
 import org.apache.doris.fs.operations.HDFSFileOperations;
 import org.apache.doris.fs.operations.HDFSOpParams;
 import org.apache.doris.fs.operations.OpParams;
+import org.apache.doris.fs.remote.RemoteFSPhantomManager;
 import org.apache.doris.fs.remote.RemoteFile;
 import org.apache.doris.fs.remote.RemoteFileSystem;
 
@@ -75,8 +76,14 @@ public DFSFileSystem(StorageBackend.StorageType type, Map<String, String> proper
     @VisibleForTesting
     @Override
     public FileSystem nativeFileSystem(String remotePath) throws UserException {
+        if (closed.get()) {
+            throw new UserException("FileSystem is closed.");
+        }
         if (dfsFileSystem == null) {
             synchronized (this) {
+                if (closed.get()) {
+                    throw new UserException("FileSystem is closed.");
+                }
                 if (dfsFileSystem == null) {
                     Configuration conf = getHdfsConf(ifNotSetFallbackToSimpleAuth());
                     for (Map.Entry<String, String> propEntry : properties.entrySet()) {
@@ -96,6 +103,7 @@ public FileSystem nativeFileSystem(String remotePath) throws UserException {
                         throw new UserException(e);
                     }
                     operations = new HDFSFileOperations(dfsFileSystem);
+                    RemoteFSPhantomManager.registerPhantomReference(this);
                 }
             }
         }