[fix](fe) Keep cached file systems alive while in use (#63677)

### What problem does this PR solve?

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

Hive file listing and Hive ACID state loading borrow `FileSystem`
instances from `FileSystemCache` and keep using them while checking
splitability, listing remote files, or loading ACID state.

The previous cache implementation returned the raw cached `FileSystem`.
When a cache entry was evicted or expired, the Caffeine removal listener
closed that same instance immediately. If cache cleanup happened while
another thread was still using the returned instance, the active Hive
operation could observe a closed filesystem.

This PR fixes the lifecycle in `FileSystemCache` instead of bypassing
the cache at Hive call sites. Cached filesystems are now returned
through leases backed by a holder with an active reference count. Cache
eviction marks the holder as evicted, and the underlying filesystem is
closed only after the last active lease is released. If the filesystem
cache is disabled, the direct lease owns the newly created filesystem
and closes it when released. Hive file listing and ACID paths now use
try-with-resources to hold the lease for the whole filesystem usage
window.

### Release note

None

### Check List (For Author)

- Test <!-- At least one of them must be included. -->
    - [ ] Regression test
    - [x] Unit Test
- `./run-fe-ut.sh --run
org.apache.doris.fs.FileSystemCacheTest,org.apache.doris.datasource.hive.HiveMetaStoreCacheTest`
    - [ ] Manual test (add detailed scripts or steps below)
    - [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
        - [ ] Previous test can cover this change.
        - [ ] No code files have been changed.
        - [ ] Other reason <!-- Add your reason?  -->

- Behavior changed:
    - [x] No.
    - [ ] Yes. <!-- Explain the behavior change -->

- Does this need documentation?
    - [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
apache/doris-website#1214 -->

### Check List (For Reviewer who merge this PR)

- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->

Author: suxiaogang223

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveExternalMetaCache.java

@@ -394,13 +394,13 @@ private FileCacheValue getFileCache(HMSExternalCatalog catalog,
 
         FileSystemCache.FileSystemCacheKey fileSystemCacheKey = new FileSystemCache.FileSystemCacheKey(
                 path.getFsIdentifier(), path.getStorageProperties());
-        FileSystem fs = Env.getCurrentEnv().getExtMetaCacheMgr().getFsCache()
-                .getFileSystem(fileSystemCacheKey);
-        result.setSplittable(HiveUtil.isSplittable(fs, inputFormat, path.getNormalizedLocation()));
 
         boolean isRecursiveDirectories = Boolean.valueOf(
                 catalog.getProperties().getOrDefault("hive.recursive_directories", "true"));
-        try {
+        try (FileSystemCache.FileSystemLease fileSystemLease = Env.getCurrentEnv().getExtMetaCacheMgr().getFsCache()
+                .getFileSystem(fileSystemCacheKey)) {
+            FileSystem fs = fileSystemLease.fileSystem();
+            result.setSplittable(HiveUtil.isSplittable(fs, inputFormat, path.getNormalizedLocation()));
             RemoteIterator<FileEntry> iterator = directoryLister.listFiles(fs, isRecursiveDirectories,
                     table, path.getNormalizedLocation());
             boolean isLzoInputFormat = HiveUtil.isLzoInputFormat(inputFormat);
@@ -824,22 +824,24 @@ public List<FileCacheValue> getFilesByTransaction(List<HivePartition> partitions
                 HMSExternalCatalog catalog = hmsCatalog(partition.getNameMapping().getCtlId());
                 LocationPath locationPath = LocationPath.of(partition.getPath(),
                         catalog.getCatalogProperty().getStoragePropertiesMap());
-                FileSystem fileSystem = Env.getCurrentEnv().getExtMetaCacheMgr().getFsCache()
-                        .getFileSystem(new FileSystemCache.FileSystemCacheKey(
-                                locationPath.getNormalizedLocation(),
-                                locationPath.getStorageProperties()));
+                FileSystemCache.FileSystemCacheKey fileSystemCacheKey = new FileSystemCache.FileSystemCacheKey(
+                        locationPath.getNormalizedLocation(), locationPath.getStorageProperties());
                 AuthenticationConfig authenticationConfig = AuthenticationConfig
                         .getKerberosConfig(locationPath.getStorageProperties().getBackendConfigProperties());
                 HadoopAuthenticator hadoopAuthenticator =
                         HadoopAuthenticator.getHadoopAuthenticator(authenticationConfig);
 
-                fileCacheValues.add(
-                        hadoopAuthenticator.doAs(() -> AcidUtil.getAcidState(
-                                fileSystem,
-                                partition,
-                                txnValidIds,
-                                catalog.getCatalogProperty().getStoragePropertiesMap(),
-                                isFullAcid)));
+                try (FileSystemCache.FileSystemLease fileSystemLease =
+                        Env.getCurrentEnv().getExtMetaCacheMgr().getFsCache().getFileSystem(fileSystemCacheKey)) {
+                    FileSystem fileSystem = fileSystemLease.fileSystem();
+                    fileCacheValues.add(
+                            hadoopAuthenticator.doAs(() -> AcidUtil.getAcidState(
+                                    fileSystem,
+                                    partition,
+                                    txnValidIds,
+                                    catalog.getCatalogProperty().getStoragePropertiesMap(),
+                                    isFullAcid)));
+                }
             }
         } catch (Exception e) {
             throw new CacheException("Failed to get input splits %s", e, txnValidIds.toString());

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

@@ -23,48 +23,185 @@
 import org.apache.doris.filesystem.FileSystem;
 
 import com.github.benmanes.caffeine.cache.LoadingCache;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
 import java.util.Objects;
 import java.util.OptionalLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 
 public class FileSystemCache {
 
     private static final Logger LOG = LogManager.getLogger(FileSystemCache.class);
 
-    private final LoadingCache<FileSystemCacheKey, FileSystem> fileSystemCache;
+    private final LoadingCache<FileSystemCacheKey, FileSystemHolder> fileSystemCache;
+    private final Function<FileSystemCacheKey, FileSystem> loader;
 
     public FileSystemCache() {
+        this(
+                Config.max_remote_file_system_cache_num,
+                OptionalLong.of(Config.external_cache_expire_time_seconds_after_access),
+                FileSystemCache::loadFileSystem);
+    }
+
+    @VisibleForTesting
+    FileSystemCache(long maxSize, OptionalLong expireAfterAccessSec, Function<FileSystemCacheKey, FileSystem> loader) {
+        this.loader = Objects.requireNonNull(loader, "loader");
+        if (maxSize == 0) {
+            fileSystemCache = null;
+            return;
+        }
         // no need to set refreshAfterWrite, because the FileSystem is created once and never changed
         CacheFactory fsCacheFactory = new CacheFactory(
-                OptionalLong.of(Config.external_cache_expire_time_seconds_after_access),
+                expireAfterAccessSec,
                 OptionalLong.empty(),
-                Config.max_remote_file_system_cache_num,
+                maxSize,
                 false,
                 null);
-        fileSystemCache = fsCacheFactory.buildCacheWithSyncRemovalListener(this::loadFileSystem, (key, fs, cause) -> {
-            if (fs != null) {
-                try {
-                    fs.close();
-                } catch (IOException e) {
-                    LOG.warn("Failed to close evicted FileSystem for key: {}", key, e);
-                }
-            }
-        });
+        fileSystemCache = fsCacheFactory.buildCacheWithSyncRemovalListener(
+                key -> new FileSystemHolder(key, loader.apply(key)), (key, holder, cause) -> {
+                    if (holder != null) {
+                        holder.markEvicted();
+                    }
+                });
     }
 
-    private FileSystem loadFileSystem(FileSystemCacheKey key) {
+    private static FileSystem loadFileSystem(FileSystemCacheKey key) {
         try {
             return FileSystemFactory.getFileSystem(key.properties);
         } catch (IOException e) {
             throw new RuntimeException("Failed to create filesystem for key: " + key, e);
         }
     }
 
-    public FileSystem getFileSystem(FileSystemCacheKey key) {
-        return fileSystemCache.get(key);
+    public FileSystemLease getFileSystem(FileSystemCacheKey key) {
+        if (fileSystemCache == null) {
+            return new DirectFileSystemLease(key, loader.apply(key));
+        }
+        while (true) {
+            FileSystemHolder holder = fileSystemCache.get(key);
+            FileSystemLease lease = holder.acquire();
+            if (lease != null) {
+                return lease;
+            }
+            fileSystemCache.asMap().remove(key, holder);
+        }
+    }
+
+    @VisibleForTesting
+    void cleanUp() {
+        if (fileSystemCache != null) {
+            fileSystemCache.cleanUp();
+        }
+    }
+
+    private static final class FileSystemHolder {
+        private final FileSystemCacheKey key;
+        private final FileSystem fileSystem;
+        private int referenceCount = 0;
+        private boolean evicted = false;
+        private boolean closed = false;
+
+        private FileSystemHolder(FileSystemCacheKey key, FileSystem fileSystem) {
+            this.key = Objects.requireNonNull(key, "key");
+            this.fileSystem = Objects.requireNonNull(fileSystem, "fileSystem");
+        }
+
+        private synchronized FileSystemLease acquire() {
+            if (evicted || closed) {
+                return null;
+            }
+            referenceCount++;
+            return new CachedFileSystemLease(this);
+        }
+
+        private synchronized void release() {
+            Preconditions.checkState(referenceCount > 0, "FileSystem lease has been released more than once");
+            referenceCount--;
+            closeIfIdle();
+        }
+
+        private synchronized void markEvicted() {
+            evicted = true;
+            closeIfIdle();
+        }
+
+        private void closeIfIdle() {
+            if (!evicted || referenceCount != 0 || closed) {
+                return;
+            }
+            closed = true;
+            try {
+                fileSystem.close();
+            } catch (IOException e) {
+                LOG.warn("Failed to close evicted FileSystem for key: {}", key, e);
+            }
+        }
+
+        private FileSystem fileSystem() {
+            return fileSystem;
+        }
+    }
+
+    public interface FileSystemLease extends AutoCloseable {
+        FileSystem fileSystem();
+
+        @Override
+        void close();
+    }
+
+    private static final class CachedFileSystemLease implements FileSystemLease {
+        private final FileSystemHolder holder;
+        private final AtomicBoolean closed = new AtomicBoolean(false);
+
+        private CachedFileSystemLease(FileSystemHolder holder) {
+            this.holder = holder;
+        }
+
+        @Override
+        public FileSystem fileSystem() {
+            return holder.fileSystem();
+        }
+
+        @Override
+        public void close() {
+            if (!closed.compareAndSet(false, true)) {
+                return;
+            }
+            holder.release();
+        }
+    }
+
+    private static final class DirectFileSystemLease implements FileSystemLease {
+        private final FileSystemCacheKey key;
+        private final FileSystem fileSystem;
+        private final AtomicBoolean closed = new AtomicBoolean(false);
+
+        private DirectFileSystemLease(FileSystemCacheKey key, FileSystem fileSystem) {
+            this.key = key;
+            this.fileSystem = fileSystem;
+        }
+
+        @Override
+        public FileSystem fileSystem() {
+            return fileSystem;
+        }
+
+        @Override
+        public void close() {
+            if (!closed.compareAndSet(false, true)) {
+                return;
+            }
+            try {
+                fileSystem.close();
+            } catch (IOException e) {
+                LOG.warn("Failed to close uncached FileSystem for key: {}", key, e);
+            }
+        }
     }
 
     public static class FileSystemCacheKey {

fe/fe-core/src/test/java/org/apache/doris/fs/FileSystemCacheTest.java

@@ -0,0 +1,139 @@
+// 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;
+
+import org.apache.doris.datasource.property.storage.StorageProperties;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.OptionalLong;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class FileSystemCacheTest {
+
+    @Test
+    public void testEvictedFileSystemClosesAfterLastLeaseIsReleased() {
+        CountingFileSystem first = new CountingFileSystem();
+        CountingFileSystem second = new CountingFileSystem();
+        AtomicInteger loadCount = new AtomicInteger();
+        FileSystemCache cache = new FileSystemCache(1L, OptionalLong.empty(),
+                key -> loadCount.getAndIncrement() == 0 ? first : second);
+
+        FileSystemCache.FileSystemCacheKey firstKey = key("hdfs://ns1");
+        FileSystemCache.FileSystemLease firstLease = cache.getFileSystem(firstKey);
+        Assert.assertSame(first, firstLease.fileSystem());
+
+        FileSystemCache.FileSystemLease secondLease = cache.getFileSystem(key("hdfs://ns2"));
+        cache.cleanUp();
+
+        Assert.assertEquals(0, first.getCloseCount());
+        Assert.assertEquals(0, second.getCloseCount());
+
+        firstLease.close();
+        Assert.assertEquals(1, first.getCloseCount());
+        Assert.assertEquals(0, second.getCloseCount());
+
+        secondLease.close();
+        Assert.assertEquals(0, second.getCloseCount());
+    }
+
+    @Test
+    public void testUncachedFileSystemClosesWhenLeaseIsReleased() {
+        CountingFileSystem fileSystem = new CountingFileSystem();
+        FileSystemCache cache = new FileSystemCache(0L, OptionalLong.empty(), key -> fileSystem);
+
+        FileSystemCache.FileSystemLease lease = cache.getFileSystem(key("hdfs://ns1"));
+        Assert.assertSame(fileSystem, lease.fileSystem());
+        Assert.assertEquals(0, fileSystem.getCloseCount());
+
+        lease.close();
+        Assert.assertEquals(1, fileSystem.getCloseCount());
+    }
+
+    @Test
+    public void testLeaseCloseIsConcurrentIdempotent() throws InterruptedException {
+        CountingFileSystem fileSystem = new CountingFileSystem();
+        FileSystemCache cache = new FileSystemCache(1L, OptionalLong.empty(), key -> fileSystem);
+
+        FileSystemCache.FileSystemLease lease = cache.getFileSystem(key("hdfs://ns1"));
+        FileSystemCache.FileSystemLease evictingLease = cache.getFileSystem(key("hdfs://ns2"));
+        cache.cleanUp();
+
+        closeConcurrently(lease);
+        Assert.assertEquals(1, fileSystem.getCloseCount());
+
+        evictingLease.close();
+    }
+
+    @Test
+    public void testUncachedLeaseCloseIsConcurrentIdempotent() throws InterruptedException {
+        CountingFileSystem fileSystem = new CountingFileSystem();
+        FileSystemCache cache = new FileSystemCache(0L, OptionalLong.empty(), key -> fileSystem);
+
+        closeConcurrently(cache.getFileSystem(key("hdfs://ns1")));
+
+        Assert.assertEquals(1, fileSystem.getCloseCount());
+    }
+
+    private static void closeConcurrently(FileSystemCache.FileSystemLease lease) throws InterruptedException {
+        CountDownLatch ready = new CountDownLatch(2);
+        CountDownLatch start = new CountDownLatch(1);
+        Thread first = closeThread(lease, ready, start);
+        Thread second = closeThread(lease, ready, start);
+        first.start();
+        second.start();
+        ready.await();
+        start.countDown();
+        first.join();
+        second.join();
+    }
+
+    private static Thread closeThread(FileSystemCache.FileSystemLease lease,
+            CountDownLatch ready, CountDownLatch start) {
+        return new Thread(() -> {
+            ready.countDown();
+            try {
+                start.await();
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+            lease.close();
+        });
+    }
+
+    private static FileSystemCache.FileSystemCacheKey key(String fsIdent) {
+        return new FileSystemCache.FileSystemCacheKey(fsIdent, StorageProperties.createPrimary(
+                Collections.singletonMap(StorageProperties.FS_HDFS_SUPPORT, "true")));
+    }
+
+    private static class CountingFileSystem extends MemoryFileSystem {
+        private final AtomicInteger closeCount = new AtomicInteger();
+
+        @Override
+        public void close() {
+            closeCount.incrementAndGet();
+        }
+
+        int getCloseCount() {
+            return closeCount.get();
+        }
+    }
+}