diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java index 2b0dcec3f760..846859ed2e40 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java +++ b/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. + * + *

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; } } diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java index 587c1f2d4423..3acbbbb574bd 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java +++ b/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); diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java index 93bc4a4f6f90..bf7e555e33df 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/ResolvingFileIO.java +++ b/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 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; }); } diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java index 3ff241d6c8f2..35b7b48ca226 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java +++ b/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, 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 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. + * + *

This mirrors the branch Hadoop itself takes in {@code FileSystem#get(URI, Configuration)}: + * with {@code fs..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. + * + *

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 owned = new ArrayList<>(); + synchronized (this) { + closed = true; + Map, FileSystem> map = fsMap; + if (map == null) { + return; + } + for (Map.Entry, 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 { /** diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java b/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java index cbfca1b6d953..a5d7ba7d182f 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystem.java +++ b/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( () -> { diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java index 2411cce3d030..08eea3e4169a 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/PluginFileIOTest.java @@ -26,7 +26,11 @@ import java.time.Duration; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** Tests for {@link PluginFileIO}. */ @@ -56,11 +60,59 @@ void testCreateBlobPresignedUrlUsesPluginClassLoader() throws IOException { assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(original); } + @Test + void testCloseReleasesTheDelegateUnderThePluginClassLoader() throws IOException { + FileIO delegate = mock(FileIO.class); + ClassLoader pluginClassLoader = new ClassLoader() {}; + TestPluginFileIO fileIO = new TestPluginFileIO(delegate, pluginClassLoader); + ClassLoader original = Thread.currentThread().getContextClassLoader(); + doAnswer( + ignored -> { + assertThat(Thread.currentThread().getContextClassLoader()) + .isSameAs(pluginClassLoader); + return null; + }) + .when(delegate) + .close(); + + // nothing has been resolved yet, so there is nothing to release + new TestPluginFileIO(delegate, pluginClassLoader).close(); + verify(delegate, never()).close(); + + fileIO.exists(new Path("oss://bucket/table/file")); + fileIO.close(); + + verify(delegate).close(); + assertThat(Thread.currentThread().getContextClassLoader()).isSameAs(original); + + // the delegate has been dropped, closing again must not touch it a second time + fileIO.close(); + verify(delegate).close(); + } + + @Test + void testUseAfterCloseFailsWithIOExceptionRatherThanNpe() throws IOException { + FileIO delegate = mock(FileIO.class); + TestPluginFileIO fileIO = new TestPluginFileIO(delegate, new ClassLoader() {}); + Path path = new Path("oss://bucket/table/file"); + + fileIO.exists(path); + fileIO.close(); + + // close() is the only writer of null, so a plain field read here would NPE, and silently + // re-creating the delegate would build a second file system nobody releases + assertThatThrownBy(() -> fileIO.exists(path)) + .isInstanceOf(IOException.class) + .hasMessageContaining("closed"); + assertThat(fileIO.createdCount).isEqualTo(1); + } + private static class TestPluginFileIO extends PluginFileIO { private final FileIO delegate; private final ClassLoader classLoader; private Path createdFor; + private int createdCount; private TestPluginFileIO(FileIO delegate, ClassLoader classLoader) { this.delegate = delegate; @@ -75,6 +127,7 @@ public boolean isObjectStore() { @Override protected FileIO createFileIO(Path path) { createdFor = path; + createdCount++; return delegate; } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java index c550b84df171..8ba88b2d6da4 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/ResolvingFileIOTest.java @@ -29,15 +29,21 @@ import java.io.IOException; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -144,6 +150,79 @@ public void testFileIOMapStoresFileIOInstances() throws IOException { assertEquals(hdfsFileIO, hdfsFileIOAgain); } + @Test + public void testCloseReleasesEveryResolvedFileIO() throws Exception { + List loaded = new ArrayList<>(); + configureWithFreshDelegates(loaded, false); + + // two authorities mean two entries in the delegate map + FileIO first = resolvingFileIO.fileIO(new Path("oss://bucket-1/table")); + FileIO second = resolvingFileIO.fileIO(new Path("oss://bucket-2/table")); + assertNotEquals(first, second); + + resolvingFileIO.close(); + + // both resolved delegates, and the throwaway probes FileIO.get made along the way + for (FileIO fileIO : loaded) { + verify(fileIO, times(1)).close(); + } + } + + @Test + public void testCloseKeepsGoingWhenADelegateFails() throws Exception { + List loaded = new ArrayList<>(); + configureWithFreshDelegates(loaded, true); + + FileIO first = resolvingFileIO.fileIO(new Path("oss://bucket-1/table")); + FileIO second = resolvingFileIO.fileIO(new Path("oss://bucket-2/table")); + + // the first failure must not keep the second entry from being closed + assertThrows(IOException.class, () -> resolvingFileIO.close()); + + verify(first, times(1)).close(); + verify(second, times(1)).close(); + } + + @Test + public void testUseAfterCloseIsRejectedAndTheMapIsEmptied() throws Exception { + List loaded = new ArrayList<>(); + configureWithFreshDelegates(loaded, false); + + FileIO delegate = resolvingFileIO.fileIO(new Path("oss://bucket-1/table")); + resolvingFileIO.close(); + + // the map really is drained, so a second close must not close the delegate again + resolvingFileIO.close(); + verify(delegate, times(1)).close(); + + // and resolving again must not silently rebuild a delegate nobody will release + assertThrows( + IOException.class, () -> resolvingFileIO.fileIO(new Path("oss://bucket-1/table"))); + } + + /** + * Hands out a fresh delegate per load, the way a real loader does. FileIO.get loads one + * throwaway instance to probe access and another one to return, so a single shared mock cannot + * tell the two apart. + */ + private void configureWithFreshDelegates(List loaded, boolean failOnClose) + throws IOException { + FileIOLoader loader = mock(FileIOLoader.class); + when(loader.getScheme()).thenReturn("oss"); + when(loader.load(any())) + .thenAnswer( + ignored -> { + FileIO delegate = mock(FileIO.class); + when(delegate.exists(any())).thenReturn(true); + if (failOnClose) { + doThrow(new IOException("cannot close")).when(delegate).close(); + } + loaded.add(delegate); + return delegate; + }); + resolvingFileIO.configure(CatalogContext.create(new Options(), loader, null)); + } + @Test public void testCreateBlobPresignedUrlResolvesDescriptorFileIO() throws IOException { FileIO delegate = mock(FileIO.class); diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOTest.java new file mode 100644 index 000000000000..2db9ad0b55bf --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOTest.java @@ -0,0 +1,487 @@ +/* + * 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.paimon.fs.hadoop; + +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.options.Options; +import org.apache.paimon.utils.Pair; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.util.Progressable; +import org.junit.jupiter.api.Test; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.net.URI; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Tests for {@link HadoopFileIO}, mostly around releasing the file systems it owns. */ +public class HadoopFileIOTest { + + @Test + public void testUncachedFileSystemIsClosed() throws Exception { + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://owned/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://owned/a"); + assertThat(fs.closeCount()).isZero(); + + fileIO.close(); + + assertThat(fs.closeCount()).isEqualTo(1); + assertThat(fileIO.fsMap).isEmpty(); + } + + @Test + public void testCachedFileSystemIsNotClosed() throws Exception { + // without disable.cache the instance comes from Hadoop's global cache and is shared with + // every other user in this JVM, so closing it here would break unrelated readers + Configuration conf = conf("testfs"); + HadoopFileIO fileIO = fileIO(conf, "testfs://shared/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://shared/a"); + try { + fileIO.close(); + + assertThat(fs.closeCount()).isZero(); + + // still shared: another FileIO gets the very same instance back + HadoopFileIO other = fileIO(conf, "testfs://shared/warehouse"); + assertThat(fileSystem(other, "testfs://shared/a")).isSameAs(fs); + } finally { + // Hadoop's cache is static and this fork is reused, so never leave it behind + fs.close(); + } + } + + @Test + public void testOnlyTheOwnedSchemeIsClosed() throws Exception { + Configuration conf = conf("testfs", "otherfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://mixed/warehouse"); + + RecordingFileSystem owned = fileSystem(fileIO, "testfs://mixed/a"); + RecordingFileSystem shared = fileSystem(fileIO, "otherfs://mixed/a"); + + try { + fileIO.close(); + + assertThat(owned.closeCount()).isEqualTo(1); + assertThat(shared.closeCount()).isZero(); + } finally { + shared.close(); + } + } + + @Test + public void testFailingCloseDoesNotSkipTheOtherFileSystems() throws Exception { + Configuration conf = conf("testfs", "badfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + conf.setBoolean("fs.badfs.impl.disable.cache", true); + conf.setBoolean("fs.badfs.test.fail-on-close", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://failing/warehouse"); + + RecordingFileSystem bad = fileSystem(fileIO, "badfs://failing/a"); + RecordingFileSystem good = fileSystem(fileIO, "testfs://failing/a"); + + assertThatThrownBy(fileIO::close).isInstanceOf(IOException.class); + + assertThat(bad.closeCount()).isEqualTo(1); + assertThat(good.closeCount()).isEqualTo(1); + assertThat(fileIO.fsMap).isEmpty(); + } + + @Test + public void testCloseIsIdempotent() throws Exception { + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://idempotent/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://idempotent/a"); + fileIO.close(); + fileIO.close(); + + assertThat(fs.closeCount()).isEqualTo(1); + } + + @Test + public void testCloseBeforeAnyUseIsSafe() throws Exception { + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + // nothing has been opened, so the lazily created map is still null; this is also the state + // of every instance that arrives by deserialization + HadoopFileIO fileIO = fileIO(conf, "testfs://untouched/warehouse"); + + fileIO.close(); + fileIO.close(); + } + + @Test + public void testInjectedFileSystemIsNotClosed() throws Exception { + // an unconfigured FileIO owns nothing, and a file system handed in from the outside stays + // the caller's to close + HadoopFileIO fileIO = new HadoopFileIO(new Path("testfs://injected/warehouse")); + RecordingFileSystem fs = new RecordingFileSystem(); + fileIO.setFileSystem(fs); + + fileIO.close(); + + assertThat(fs.closeCount()).isZero(); + } + + @Test + public void testFileSystemLosingTheCreationRaceIsReleased() throws Exception { + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + RecordingFileSystem winner = new RecordingFileSystem(); + RacingHadoopFileIO fileIO = + new RacingHadoopFileIO(new Path("testfs://race/warehouse"), winner); + fileIO.configure(CatalogContext.create(new Options(), conf)); + + FileSystem returned = + fileIO.getFileSystem(new org.apache.hadoop.fs.Path("testfs://race/a")); + + assertThat(returned).isSameAs(winner); + assertThat(fileIO.loser.closeCount()).isEqualTo(1); + assertThat(winner.closeCount()).isZero(); + } + + @Test + public void testOwnershipUsesTheSchemeAsWritten() { + Configuration conf = new Configuration(); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://spelling/warehouse"); + + assertThat(fileIO.isOwnedScheme("testfs")).isTrue(); + // Hadoop looks the property up with the scheme exactly as the path spells it, so a + // differently cased scheme is served from the global cache and is not ours to close + assertThat(fileIO.isOwnedScheme("TESTFS")).isFalse(); + assertThat(fileIO.isOwnedScheme("otherfs")).isFalse(); + } + + @Test + public void testSchemelessPathFallsBackToTheDefaultFileSystem() { + Configuration conf = new Configuration(); + conf.set("fs.defaultFS", "testfs://default"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://default/warehouse"); + + assertThat(fileIO.isOwnedScheme(null)).isTrue(); + + conf.setBoolean("fs.testfs.impl.disable.cache", false); + assertThat(fileIO.isOwnedScheme(null)).isFalse(); + } + + @Test + public void testUseAfterCloseIsRejectedInsteadOfLeakingAgain() throws Exception { + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://reuse/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://reuse/a"); + fileIO.close(); + assertThat(fs.closeCount()).isEqualTo(1); + + // resurrecting would silently create an owned file system that nobody will ever release + RecordingFileSystem.resetCounters(); + assertThatThrownBy(() -> fileSystem(fileIO, "testfs://reuse/a")) + .isInstanceOf(IOException.class) + .hasMessageContaining("closed"); + assertThat(RecordingFileSystem.created()).isZero(); + } + + @Test + public void testFileSystemCreatedWhileClosingIsReleased() throws Exception { + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + // stands in for close() landing between the creation and the publication of a file system + ClosingWhileCreatingFileIO fileIO = + new ClosingWhileCreatingFileIO(new Path("testfs://late/warehouse")); + fileIO.configure(CatalogContext.create(new Options(), conf)); + + assertThatThrownBy( + () -> + fileIO.getFileSystem( + new org.apache.hadoop.fs.Path("testfs://late/a"))) + .isInstanceOf(IOException.class) + .hasMessageContaining("closed"); + + assertThat(fileIO.created.closeCount()).isEqualTo(1); + assertThat(fileIO.fsMap).isEmpty(); + } + + @Test + public void testCachedFileSystemLosingTheRaceIsNotClosed() throws Exception { + // with the Hadoop cache enabled both racing threads get the very same shared instance, so + // releasing the loser would close the instance we are about to hand out + Configuration conf = conf("testfs"); + RacingHadoopFileIO fileIO = + new RacingHadoopFileIO(new Path("testfs://cachedrace/warehouse"), null); + fileIO.configure(CatalogContext.create(new Options(), conf)); + + FileSystem returned = + fileIO.getFileSystem(new org.apache.hadoop.fs.Path("testfs://cachedrace/a")); + + assertThat(((RecordingFileSystem) returned).closeCount()).isZero(); + ((RecordingFileSystem) returned).close(); + } + + @Test + public void testOwnershipFollowsThePathSchemeNotTheFileSystemUri() throws Exception { + // the file system reports a scheme of its own; ownership must still follow the path, which + // is the scheme Hadoop consulted when it decided whether to cache + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + conf.set("fs.testfs.test.reported-scheme", "otherfs"); + HadoopFileIO fileIO = fileIO(conf, "testfs://reported/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://reported/a"); + assertThat(fs.getUri().getScheme()).isEqualTo("otherfs"); + + fileIO.close(); + + assertThat(fs.closeCount()).isEqualTo(1); + } + + @Test + public void testMalformedDefaultFileSystemDoesNotBreakClose() throws Exception { + // Hadoop 3 throws IllegalArgumentException out of getDefaultUri for a scheme-less + // fs.defaultFS, which must not escape close() and strand everything else + Configuration conf = new Configuration(); + conf.set("fs.defaultFS", "no-scheme-here"); + HadoopFileIO fileIO = fileIO(conf, "testfs://malformed/warehouse"); + + assertThat(fileIO.isOwnedScheme(null)).isFalse(); + fileIO.close(); + } + + @Test + public void testAccessProbeFileSystemIsNotLeakedByFileIOGet() throws Exception { + // FileIO.get probes a loader by calling exists() on a throwaway FileIO; with the Hadoop + // cache disabled that probe creates a file system nobody else can reach + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + RecordingFileSystem.resetCounters(); + + Path path = new Path("testfs://probe/warehouse"); + FileIO fileIO = FileIO.get(path, CatalogContext.create(new Options(), conf)); + try { + fileIO.exists(path); + } finally { + fileIO.close(); + } + + assertThat(RecordingFileSystem.created()).isGreaterThan(1); + assertThat(RecordingFileSystem.closed()).isEqualTo(RecordingFileSystem.created()); + } + + private static Configuration conf(String... schemes) { + Configuration conf = new Configuration(); + for (String scheme : schemes) { + conf.set("fs." + scheme + ".impl", RecordingFileSystem.class.getName()); + } + return conf; + } + + private static HadoopFileIO fileIO(Configuration conf, String warehouse) { + HadoopFileIO fileIO = new HadoopFileIO(new Path(warehouse)); + fileIO.configure(CatalogContext.create(new Options(), conf)); + return fileIO; + } + + private static RecordingFileSystem fileSystem(HadoopFileIO fileIO, String path) + throws IOException { + return (RecordingFileSystem) fileIO.getFileSystem(new org.apache.hadoop.fs.Path(path)); + } + + /** + * A {@link HadoopFileIO} that always loses the race for publishing a new file system. A null + * winner publishes the created instance itself, which is what Hadoop's own cache hands to both + * racing threads. + */ + private static class RacingHadoopFileIO extends HadoopFileIO { + + private static final long serialVersionUID = 1L; + + private final FileSystem winner; + private RecordingFileSystem loser; + + private RacingHadoopFileIO(Path path, FileSystem winner) { + super(path); + this.winner = winner; + } + + @Override + protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) throws IOException { + loser = (RecordingFileSystem) super.createFileSystem(path); + // stand in for a concurrent thread that published its own instance first + URI uri = path.toUri(); + fsMap.put( + Pair.of(uri.getScheme(), uri.getAuthority()), winner == null ? loser : winner); + return loser; + } + } + + /** A {@link HadoopFileIO} that is closed in the window between creating and publishing. */ + private static class ClosingWhileCreatingFileIO extends HadoopFileIO { + + private static final long serialVersionUID = 1L; + + private RecordingFileSystem created; + + private ClosingWhileCreatingFileIO(Path path) { + super(path); + } + + @Override + protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) throws IOException { + created = (RecordingFileSystem) super.createFileSystem(path); + close(); + return created; + } + } + + /** A {@link FileSystem} that records how often it was closed and does nothing else. */ + public static class RecordingFileSystem extends FileSystem { + + private static final AtomicInteger CREATED = new AtomicInteger(); + private static final AtomicInteger CLOSED = new AtomicInteger(); + + private URI uri; + private boolean failOnClose; + private int closeCount; + + static void resetCounters() { + CREATED.set(0); + CLOSED.set(0); + } + + static int created() { + return CREATED.get(); + } + + static int closed() { + return CLOSED.get(); + } + + @Override + public void initialize(URI name, Configuration conf) throws IOException { + super.initialize(name, conf); + String reportedScheme = + conf.get("fs." + name.getScheme() + ".test.reported-scheme", null); + this.uri = + reportedScheme == null + ? name + : URI.create(reportedScheme + "://" + name.getAuthority()); + this.failOnClose = + conf.getBoolean("fs." + name.getScheme() + ".test.fail-on-close", false); + CREATED.incrementAndGet(); + } + + @Override + public URI getUri() { + return uri; + } + + @Override + public void close() throws IOException { + closeCount++; + CLOSED.incrementAndGet(); + super.close(); + if (failOnClose) { + throw new IOException("close fails on purpose for " + uri); + } + } + + int closeCount() { + return closeCount; + } + + @Override + public FSDataInputStream open(org.apache.hadoop.fs.Path f, int bufferSize) { + throw new UnsupportedOperationException(); + } + + @Override + public FSDataOutputStream create( + org.apache.hadoop.fs.Path f, + FsPermission permission, + boolean overwrite, + int bufferSize, + short replication, + long blockSize, + Progressable progress) { + throw new UnsupportedOperationException(); + } + + @Override + public FSDataOutputStream append( + org.apache.hadoop.fs.Path f, int bufferSize, Progressable progress) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean rename(org.apache.hadoop.fs.Path src, org.apache.hadoop.fs.Path dst) { + throw new UnsupportedOperationException(); + } + + @Override + public boolean delete(org.apache.hadoop.fs.Path f, boolean recursive) { + throw new UnsupportedOperationException(); + } + + @Override + public FileStatus[] listStatus(org.apache.hadoop.fs.Path f) { + throw new UnsupportedOperationException(); + } + + @Override + public void setWorkingDirectory(org.apache.hadoop.fs.Path dir) { + throw new UnsupportedOperationException(); + } + + @Override + public org.apache.hadoop.fs.Path getWorkingDirectory() { + return new org.apache.hadoop.fs.Path("/"); + } + + @Override + public boolean mkdirs(org.apache.hadoop.fs.Path f, FsPermission permission) { + throw new UnsupportedOperationException(); + } + + @Override + public FileStatus getFileStatus(org.apache.hadoop.fs.Path f) throws IOException { + // lets FileSystem#exists answer false instead of blowing up, which is what the + // FileIO.get access probe needs + throw new FileNotFoundException(f.toString()); + } + } +} diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java index 54de46dc7c56..519cd5364a26 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopSecuredFileSystemTest.java @@ -22,6 +22,8 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.options.Options; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileSystem; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -78,6 +80,25 @@ public void testReturnOriginalFileSystemWhenSecurityConfigIsIllegal() throws Exc .isNotInstanceOf(HadoopSecuredFileSystem.class); } + @Test + public void testCloseClosesTheWrappedFileSystem() throws Exception { + File keytabFile = new File(tmp.toFile(), "test-keytab.keytab"); + assertThat(keytabFile.createNewFile()).isTrue(); + + Options options = new Options(); + options.set("security.kerberos.login.principal", "test-user"); + options.set("security.kerberos.login.keytab", keytabFile.getAbsolutePath()); + + HadoopFileIOTest.RecordingFileSystem wrapped = new HadoopFileIOTest.RecordingFileSystem(); + FileSystem secured = + HadoopSecuredFileSystem.trySecureFileSystem(wrapped, options, new Configuration()); + assertThat(secured).isInstanceOf(HadoopSecuredFileSystem.class); + + secured.close(); + + assertThat(wrapped.closeCount()).isEqualTo(1); + } + private HadoopFileIO createFileIO(Options options) { HadoopFileIO fileIO = new HadoopFileIO(new Path("file:///tmp/test")); fileIO.configure(CatalogContext.create(options));