[common] Fix file system leaks when the Hadoop file system cache is disabled

paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java

@@ -270,8 +270,13 @@ default String createBlobPresignedUrl(
     }
 
     /**
-     * Override this method to empty, many FileIO implementation classes rely on static variables
-     * and do not have the ability to close them.
+     * Releases the resources this instance owns exclusively. The default is empty because many
+     * implementations hold nothing of their own, or reach their resources through static variables
+     * shared with the rest of the JVM, which they must not close.
+     *
+     * <p>Override it only for resources that belong to this instance alone, and make the override
+     * idempotent. Implementations that delegate to another {@link FileIO} should forward the call,
+     * otherwise the delegate can never be released.
      */
     @Override
     default void close() throws IOException {}
@@ -655,10 +660,16 @@ static FileIOLoader checkAccess(FileIOLoader fileIO, Path path, CatalogContext c
             return null;
         }
 
-        // check access
+        // check access, the probe is thrown away afterwards so it has to be released here: with
+        // the Hadoop file system cache disabled its exists() call creates a file system that no
+        // one else can reach
         FileIO io = fileIO.load(path);
-        io.configure(config);
-        io.exists(path);
+        try {
+            io.configure(config);
+            io.exists(path);
+        } finally {
+            IOUtils.closeQuietly(io);
+        }
         return fileIO;
     }
 }

paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java

@@ -37,6 +37,9 @@ public abstract class PluginFileIO implements FileIO, HadoopOptionsProvider {
 
     private transient volatile FileIO lazyFileIO;
 
+    /** Transient so that a deserialized copy starts out usable. */
+    private transient volatile boolean closed;
+
     @Override
     public void configure(CatalogContext context) {
         // Do not get Hadoop Configuration in CatalogOptions
@@ -108,14 +111,40 @@ public String createBlobPresignedUrl(
     }
 
     private FileIO fileIO(Path path) throws IOException {
-        if (lazyFileIO == null) {
+        // read into a local, close() may null the field at any point and callers dereference the
+        // result directly
+        FileIO fileIO = lazyFileIO;
+        if (fileIO == null) {
             synchronized (this) {
-                if (lazyFileIO == null) {
-                    lazyFileIO = wrap(() -> createFileIO(path));
+                if (closed) {
+                    throw new IOException("This FileIO is closed.");
+                }
+                fileIO = lazyFileIO;
+                if (fileIO == null) {
+                    fileIO = wrap(() -> createFileIO(path));
+                    lazyFileIO = fileIO;
                 }
             }
         }
-        return lazyFileIO;
+        return fileIO;
+    }
+
+    @Override
+    public void close() throws IOException {
+        FileIO fileIO;
+        synchronized (this) {
+            closed = true;
+            fileIO = lazyFileIO;
+            lazyFileIO = null;
+        }
+        if (fileIO != null) {
+            // the delegate lives in the plugin classloader, so close it under that classloader too
+            wrap(
+                    () -> {
+                        fileIO.close();
+                        return null;
+                    });
+        }
     }
 
     protected abstract FileIO createFileIO(Path path);

paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java

@@ -23,10 +23,13 @@
 import org.apache.paimon.data.BlobDescriptor;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
+import org.apache.paimon.utils.IOUtils;
 
 import java.io.IOException;
 import java.io.Serializable;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
@@ -45,6 +48,9 @@ public class ResolvingFileIO implements FileIO {
 
     private CatalogContext context;
 
+    /** Transient so that a deserialized copy starts out usable. */
+    private transient volatile boolean closed;
+
     // TODO, how to decide the real fileio is object store or not?
     @Override
     public boolean isObjectStore() {
@@ -120,15 +126,49 @@ public String createBlobPresignedUrl(
 
     @VisibleForTesting
     public FileIO fileIO(Path path) throws IOException {
+        if (closed) {
+            throw new IOException("This FileIO is closed.");
+        }
         CacheKey cacheKey = new CacheKey(path.toUri().getScheme(), path.toUri().getAuthority());
-        return fileIOMap.computeIfAbsent(
-                cacheKey,
-                k -> {
+        FileIO fileIO =
+                fileIOMap.computeIfAbsent(
+                        cacheKey,
+                        k -> {
+                            try {
+                                return FileIO.get(path, context);
+                            } catch (IOException e) {
+                                throw new RuntimeException(e);
+                            }
+                        });
+        if (closed) {
+            // a close() ran while we were resolving and may already have passed this key, so take
+            // the delegate back out rather than leaving it behind unclosed
+            fileIOMap.remove(cacheKey, fileIO);
+            IOUtils.closeQuietly(fileIO);
+            throw new IOException("This FileIO is closed.");
+        }
+        return fileIO;
+    }
+
+    @Override
+    public void close() throws IOException {
+        closed = true;
+        // remove before closing, so that a concurrent close does not close the same delegate twice
+        List<FileIO> toClose = new ArrayList<>();
+        for (CacheKey key : fileIOMap.keySet()) {
+            FileIO fileIO = fileIOMap.remove(key);
+            if (fileIO != null) {
+                toClose.add(fileIO);
+            }
+        }
+        wrap(
+                () -> {
                     try {
-                        return FileIO.get(path, context);
-                    } catch (IOException e) {
-                        throw new RuntimeException(e);
+                        IOUtils.closeAll(toClose);
+                    } catch (Exception e) {
+                        throw new IOException("Failed to close the resolved file IOs", e);
                     }
+                    return null;
                 });
     }
 

paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java

@@ -30,6 +30,7 @@
 import org.apache.paimon.hadoop.SerializableConfiguration;
 import org.apache.paimon.utils.FileIOUtils;
 import org.apache.paimon.utils.FunctionWithException;
+import org.apache.paimon.utils.IOUtils;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.ReflectionUtils;
 
@@ -39,12 +40,16 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Options;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.io.OutputStreamWriter;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -61,6 +66,12 @@ public class HadoopFileIO implements FileIO, HadoopOptionsProvider {
 
     protected transient volatile Map<Pair<String, String>, FileSystem> fsMap;
 
+    /**
+     * Transient so that a deserialized copy starts out usable: closing one instance must not
+     * disable the copies that were shipped to other processes.
+     */
+    private transient volatile boolean closed;
+
     private final Path path;
 
     public HadoopFileIO(Path path) {
@@ -197,6 +208,10 @@ private FileSystem getFileSystem(
             org.apache.hadoop.fs.Path path,
             FunctionWithException<org.apache.hadoop.fs.Path, FileSystem, IOException> creator)
             throws IOException {
+        if (closed) {
+            throw new IOException("This FileIO is closed.");
+        }
+
         if (fsMap == null) {
             synchronized (this) {
                 if (fsMap == null) {
@@ -214,7 +229,25 @@ private FileSystem getFileSystem(
         FileSystem fs = map.get(key);
         if (fs == null) {
             fs = creator.apply(path);
-            map.put(key, fs);
+            boolean owned = isOwnedScheme(scheme);
+            // publish under the same monitor close() takes, otherwise an instance created here can
+            // land in the map after close() drained it and then nobody would ever release it
+            synchronized (this) {
+                if (closed) {
+                    if (owned) {
+                        IOUtils.closeQuietly(fs);
+                    }
+                    throw new IOException("This FileIO is closed.");
+                }
+                FileSystem previous = map.putIfAbsent(key, fs);
+                if (previous != null) {
+                    // another thread won the race, release the instance we own and use theirs
+                    if (owned) {
+                        IOUtils.closeQuietly(fs);
+                    }
+                    fs = previous;
+                }
+            }
         }
         return fs;
     }
@@ -226,6 +259,64 @@ protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) throws IOE
         return fileSystem;
     }
 
+    /**
+     * Whether the {@link FileSystem} instances created for the given scheme belong to this {@link
+     * FileIO} exclusively, and may therefore be closed by it.
+     *
+     * <p>This mirrors the branch Hadoop itself takes in {@code FileSystem#get(URI, Configuration)}:
+     * with {@code fs.<scheme>.impl.disable.cache} set, Hadoop hands out a fresh instance that
+     * nobody else can reach, so releasing it is our responsibility. Otherwise the instance lives in
+     * Hadoop's global cache and is shared with every other user in this JVM, including other {@link
+     * FileIO}s and the compute engine itself; {@code FileSystem#closeAll} releases those on
+     * shutdown and closing one here would break unrelated readers.
+     *
+     * <p>The scheme is the one taken from the path, not from {@code FileSystem#getUri()}, and it is
+     * matched as written rather than lower cased, because that is what Hadoop looks up. Any
+     * deviation could report a cached, shared instance as owned.
+     */
+    @VisibleForTesting
+    boolean isOwnedScheme(@Nullable String scheme) {
+        if (hadoopConf == null) {
+            return false;
+        }
+        Configuration conf = hadoopConf.get();
+        if (scheme == null) {
+            // a path without a scheme is served by the default file system
+            try {
+                scheme = FileSystem.getDefaultUri(conf).getScheme();
+            } catch (IllegalArgumentException e) {
+                // a missing or malformed fs.defaultFS, so there is no scheme to claim ownership of
+                return false;
+            }
+        }
+        return conf.getBoolean(String.format("fs.%s.impl.disable.cache", scheme), false);
+    }
+
+    @Override
+    public void close() throws IOException {
+        List<FileSystem> owned = new ArrayList<>();
+        synchronized (this) {
+            closed = true;
+            Map<Pair<String, String>, FileSystem> map = fsMap;
+            if (map == null) {
+                return;
+            }
+            for (Map.Entry<Pair<String, String>, FileSystem> entry : map.entrySet()) {
+                if (isOwnedScheme(entry.getKey().getLeft())) {
+                    owned.add(entry.getValue());
+                }
+            }
+            // drop the cached instances as well, a closed one must never be handed out again
+            map.clear();
+        }
+
+        try {
+            IOUtils.closeAll(owned);
+        } catch (Exception e) {
+            throw new IOException("Failed to close the file systems owned by this FileIO", e);
+        }
+    }
+
     private static class HadoopSeekableInputStream extends SeekableInputStream {
 
         /**

paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java

@@ -21,6 +21,7 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.security.HadoopModule;
 import org.apache.paimon.security.SecurityConfiguration;
+import org.apache.paimon.utils.IOUtils;
 import org.apache.paimon.utils.StringUtils;
 
 import org.apache.hadoop.conf.Configuration;
@@ -170,6 +171,28 @@ public FileStatus getFileStatus(Path path) throws IOException {
         return runSecuredWithIOException(() -> fileSystem.getFileStatus(path));
     }
 
+    @Override
+    public void close() throws IOException {
+        // super.close() processes the delete-on-exit set, which is served by the wrapped file
+        // system, so it has to run while that one is still open. closeAll keeps going after the
+        // first failure and reports the rest as suppressed instead of dropping them.
+        try {
+            IOUtils.closeAll(super::close, this::closeWrapped);
+        } catch (IOException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new IOException("Failed to close the secured file system.", e);
+        }
+    }
+
+    private void closeWrapped() throws IOException {
+        runSecuredWithIOException(
+                () -> {
+                    fileSystem.close();
+                    return null;
+                });
+    }
+
     private void runSecured(final Runnable securedRunnable) {
         runSecured(
                 () -> {