diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java b/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java index ffe7214de4ab..5401c467da1f 100644 --- a/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java +++ b/paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java @@ -59,8 +59,12 @@ protected abstract MultiPartUploadStore multiPartUploadStore( @Override public void commit(FileIO fileIO) throws IOException { - try { - MultiPartUploadStore multiPartUploadStore = multiPartUploadStore(fileIO); + // a REST token FileIO comes out of a cache shared by the whole JVM, and an eviction closes + // the file system underneath an unleased caller, so hold a lease for the whole call. Only + // a RESTTokenFileIO handed in directly is covered; anything wrapping one is not + try (RESTTokenFileIO.Lease lease = RESTTokenFileIO.lease(fileIO)) { + MultiPartUploadStore multiPartUploadStore = + multiPartUploadStore(lease.fileIO(), targetPath()); multiPartUploadStore.completeMultipartUpload( objectName, uploadId, uploadedParts, byteLength); } catch (Exception e) { @@ -70,8 +74,9 @@ public void commit(FileIO fileIO) throws IOException { @Override public void discard(FileIO fileIO) throws IOException { - try { - MultiPartUploadStore multiPartUploadStore = multiPartUploadStore(fileIO); + try (RESTTokenFileIO.Lease lease = RESTTokenFileIO.lease(fileIO)) { + MultiPartUploadStore multiPartUploadStore = + multiPartUploadStore(lease.fileIO(), targetPath()); multiPartUploadStore.abortMultipartUpload(objectName, uploadId); } catch (Exception e) { LOG.warn("Failed to discard multipart upload with ID: {}", uploadId, e); @@ -90,12 +95,4 @@ public List uploadedParts() { @Override public void clean(FileIO fileIO) throws IOException {} - - private MultiPartUploadStore multiPartUploadStore(FileIO fileIO) throws IOException { - if (fileIO instanceof RESTTokenFileIO) { - RESTTokenFileIO restTokenFileIO = (RESTTokenFileIO) fileIO; - fileIO = restTokenFileIO.fileIO(); - } - return multiPartUploadStore(fileIO, targetPath()); - } } 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..a4cfc905434d 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 @@ -23,8 +23,8 @@ import org.apache.paimon.data.BlobDescriptor; import org.apache.paimon.fs.hadoop.HadoopFileIOLoader; import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.utils.IOUtils; -import org.apache.commons.io.IOUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -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 {} @@ -404,7 +409,7 @@ default void overwriteHintFile(Path path, String content) throws IOException { default void copyFile(Path sourcePath, Path targetPath, boolean overwrite) throws IOException { try (SeekableInputStream is = newInputStream(sourcePath); PositionOutputStream os = newOutputStream(targetPath, overwrite)) { - IOUtils.copy(is, os); + org.apache.commons.io.IOUtils.copy(is, os); } } @@ -498,132 +503,164 @@ static FileIO get(Path path, CatalogContext config) throws IOException { } FileIOLoader loader = null; + // the instance the access check already built and configured, carried through selection and + // handed back as the result, see checkAccess + FileIO checked = null; List ioExceptionList = new ArrayList<>(); + // the checked instance stays live until it is handed over, so any failure in the selection + // below, checked or not, has to release it + boolean handedOver = false; - // load preferIO - FileIOLoader preferIOLoader = config.preferIO(); try { - loader = checkAccess(preferIOLoader, path, config); - if (loader != null && LOG.isDebugEnabled()) { - LOG.debug( - "Found preferIOLoader {} with scheme {}.", - loader.getClass().getName(), - loader.getScheme()); + // load preferIO + FileIOLoader preferIOLoader = config.preferIO(); + try { + checked = checkAccess(preferIOLoader, path, config); + if (checked != null) { + loader = preferIOLoader; + if (LOG.isDebugEnabled()) { + LOG.debug( + "Found preferIOLoader {} with scheme {}.", + loader.getClass().getName(), + loader.getScheme()); + } + } + } catch (IOException ioException) { + ioExceptionList.add(ioException); } - } catch (IOException ioException) { - ioExceptionList.add(ioException); - } - if (loader == null) { - Map loaders = discoverLoaders(); - loader = loaders.get(uri.getScheme()); - if (!loaders.isEmpty() && LOG.isDebugEnabled()) { - LOG.debug( - "Discovered FileIOLoaders: {}.", - loaders.entrySet().stream() - .map( - e -> - String.format( - "{%s,%s}", - e.getKey(), - e.getValue().getClass().getName())) - .collect(Collectors.joining(","))); + if (loader == null) { + Map loaders = discoverLoaders(); + loader = loaders.get(uri.getScheme()); + if (!loaders.isEmpty() && LOG.isDebugEnabled()) { + LOG.debug( + "Discovered FileIOLoaders: {}.", + loaders.entrySet().stream() + .map( + e -> + String.format( + "{%s,%s}", + e.getKey(), + e.getValue().getClass().getName())) + .collect(Collectors.joining(","))); + } } - } - // load fallbackIO - FileIOLoader fallbackIO = config.fallbackIO(); - - if (loader != null) { - Set options = - config.options().keySet().stream() - .map(String::toLowerCase) - .collect(Collectors.toSet()); - Set missOptions = new HashSet<>(); - for (String[] keys : loader.requiredOptions()) { - boolean found = false; - for (String key : keys) { - if (options.contains(key.toLowerCase())) { - found = true; - break; + // load fallbackIO + FileIOLoader fallbackIO = config.fallbackIO(); + + if (loader != null) { + Set options = + config.options().keySet().stream() + .map(String::toLowerCase) + .collect(Collectors.toSet()); + Set missOptions = new HashSet<>(); + for (String[] keys : loader.requiredOptions()) { + boolean found = false; + for (String key : keys) { + if (options.contains(key.toLowerCase())) { + found = true; + break; + } + } + if (!found) { + missOptions.add(keys[0]); } } - if (!found) { - missOptions.add(keys[0]); + if (missOptions.size() > 0) { + IOException exception = + new IOException( + String.format( + "One or more required options are missing.\n\n" + + "Missing required options are:\n\n" + + "%s", + String.join("\n", missOptions))); + ioExceptionList.add(exception); + if (LOG.isDebugEnabled()) { + LOG.debug( + "Got {} but miss options. Will try to get fallback IO and Hadoop IO respectively.", + loader.getClass().getName()); + } + loader = null; + // this candidate is out of the running, so release whatever we checked for it + IOUtils.closeQuietly(checked); + checked = null; } } - if (missOptions.size() > 0) { - IOException exception = - new IOException( - String.format( - "One or more required options are missing.\n\n" - + "Missing required options are:\n\n" - + "%s", - String.join("\n", missOptions))); - ioExceptionList.add(exception); - if (LOG.isDebugEnabled()) { - LOG.debug( - "Got {} but miss options. Will try to get fallback IO and Hadoop IO respectively.", - loader.getClass().getName()); + + if (loader == null) { + try { + checked = checkAccess(fallbackIO, path, config); + if (checked != null) { + loader = fallbackIO; + if (LOG.isDebugEnabled()) { + LOG.debug( + "Got fallback FileIOLoader: {}.", loader.getClass().getName()); + } + } + } catch (IOException ioException) { + ioExceptionList.add(ioException); } - loader = null; } - } - if (loader == null) { - try { - loader = checkAccess(fallbackIO, path, config); - if (loader != null && LOG.isDebugEnabled()) { - LOG.debug("Got fallback FileIOLoader: {}.", loader.getClass().getName()); + // load hadoopIO + if (loader == null) { + FileIOLoader hadoopIOLoader = new HadoopFileIOLoader(); + try { + checked = checkAccess(hadoopIOLoader, path, config); + if (checked != null) { + loader = hadoopIOLoader; + if (LOG.isDebugEnabled()) { + LOG.debug("Got hadoop FileIOLoader: {}.", loader.getClass().getName()); + } + } + } catch (IOException ioException) { + ioExceptionList.add(ioException); } - } catch (IOException ioException) { - ioExceptionList.add(ioException); } - } - // load hadoopIO - if (loader == null) { - try { - loader = checkAccess(new HadoopFileIOLoader(), path, config); - if (loader != null && LOG.isDebugEnabled()) { - LOG.debug("Got hadoop FileIOLoader: {}.", loader.getClass().getName()); + if (loader == null) { + String fallbackMsg = ""; + String preferMsg = ""; + if (preferIOLoader != null) { + preferMsg = + " " + + preferIOLoader.getClass().getSimpleName() + + " also cannot access this path."; + } + if (fallbackIO != null) { + fallbackMsg = + " " + + fallbackIO.getClass().getSimpleName() + + " also cannot access this path."; + } + UnsupportedSchemeException ex = + new UnsupportedSchemeException( + String.format( + "Could not find a file io implementation for scheme '%s' in the classpath." + + "%s %s Hadoop FileSystem also cannot access this path '%s'.", + uri.getScheme(), preferMsg, fallbackMsg, path)); + for (IOException ioException : ioExceptionList) { + ex.addSuppressed(ioException); } - } catch (IOException ioException) { - ioExceptionList.add(ioException); - } - } - if (loader == null) { - String fallbackMsg = ""; - String preferMsg = ""; - if (preferIOLoader != null) { - preferMsg = - " " - + preferIOLoader.getClass().getSimpleName() - + " also cannot access this path."; + throw ex; } - if (fallbackIO != null) { - fallbackMsg = - " " - + fallbackIO.getClass().getSimpleName() - + " also cannot access this path."; - } - UnsupportedSchemeException ex = - new UnsupportedSchemeException( - String.format( - "Could not find a file io implementation for scheme '%s' in the classpath." - + "%s %s Hadoop FileSystem also cannot access this path '%s'.", - uri.getScheme(), preferMsg, fallbackMsg, path)); - for (IOException ioException : ioExceptionList) { - ex.addSuppressed(ioException); + + if (checked != null) { + // already configured by the access check + handedOver = true; + return checked; } - throw ex; + FileIO fileIO = loader.load(path); + fileIO.configure(config); + return fileIO; + } finally { + if (!handedOver) { + IOUtils.closeQuietly(checked); + } } - - FileIO fileIO = loader.load(path); - fileIO.configure(config); - return fileIO; } /** Discovers all {@link FileIOLoader} by service loader. */ @@ -649,16 +686,34 @@ static Map discoverLoaders() { return results; } - static FileIOLoader checkAccess(FileIOLoader fileIO, Path path, CatalogContext config) + /** + * Loads and configures a {@link FileIO} from the given loader and checks that it can reach the + * path. Returns the very instance it checked, or null when there is no loader. + * + *

The instance is returned rather than discarded because {@link FileIOLoader#load} is under + * no obligation to hand out a fresh one: a loader that caches or returns a singleton would give + * back the instance we just released. It also saves building the file system twice, since the + * check already built one. A failed check releases it here, and a caller that ends up rejecting + * this loader releases it there; with the Hadoop file system cache disabled that matters, + * because nobody else can reach what the check created. + */ + static FileIO checkAccess(FileIOLoader fileIO, Path path, CatalogContext config) throws IOException { if (fileIO == null) { return null; } - // check access FileIO io = fileIO.load(path); - io.configure(config); - io.exists(path); - return fileIO; + boolean accessible = false; + try { + io.configure(config); + io.exists(path); + accessible = true; + } finally { + if (!accessible) { + IOUtils.closeQuietly(io); + } + } + return io; } } 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 5568ba896cb3..35f1b5eefe28 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() { @@ -127,15 +133,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..0adb43465601 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; @@ -59,7 +64,14 @@ public class HadoopFileIO implements FileIO, HadoopOptionsProvider { private org.apache.paimon.options.Options options; - protected transient volatile Map, FileSystem> fsMap; + /** Value is the file system paired with the ownership recorded when it was created. */ + protected transient volatile Map, Pair> 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; @@ -69,7 +81,8 @@ public HadoopFileIO(Path path) { @VisibleForTesting public void setFileSystem(FileSystem fs) throws IOException { - getFileSystem(path(path), p -> fs); + // handed in from outside, so it stays the caller's to close whatever the scheme says + getFileSystem(path(path), p -> fs, false); } @Override @@ -190,13 +203,18 @@ private org.apache.hadoop.fs.Path path(Path path) { @VisibleForTesting FileSystem getFileSystem(org.apache.hadoop.fs.Path path) throws IOException { - return getFileSystem(path, this::createFileSystem); + return getFileSystem(path, this::createFileSystem, true); } private FileSystem getFileSystem( org.apache.hadoop.fs.Path path, - FunctionWithException creator) + FunctionWithException creator, + boolean mayOwn) throws IOException { + if (closed) { + throw new IOException("This FileIO is closed."); + } + if (fsMap == null) { synchronized (this) { if (fsMap == null) { @@ -205,25 +223,134 @@ private FileSystem getFileSystem( } } - Map, FileSystem> map = fsMap; + Map, Pair> map = fsMap; URI uri = path.toUri(); String scheme = uri.getScheme(); String authority = uri.getAuthority(); Pair key = Pair.of(scheme, authority); - FileSystem fs = map.get(key); - if (fs == null) { - fs = creator.apply(path); - map.put(key, fs); + Pair entry = map.get(key); + if (entry == null) { + // pin the ownership Hadoop decided on for this instance, the configuration is mutable + // and may well answer differently by the time close() asks + Pair created = + Pair.of(creator.apply(path), mayOwn && isOwnedScheme(scheme)); + entry = created; + boolean rejected = false; + // 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) { + rejected = true; + } else { + Pair previous = map.putIfAbsent(key, created); + if (previous != null) { + // another thread won the race, use theirs + entry = previous; + } + } + } + // compare the file systems, not the pairs holding them: releasing ours because the + // winner merely wrapped the same instance would close what we are about to hand out + if (rejected || entry.getLeft() != created.getLeft()) { + // outside the monitor on purpose: tearing down an object store client can block + // for a long time and everyone else needs this lock + closeIfOwned(created); + } + if (rejected) { + throw new IOException("This FileIO is closed."); + } + } + return entry.getLeft(); + } + + private static void closeIfOwned(Pair entry) { + if (entry.getRight()) { + IOUtils.closeQuietly(entry.getLeft()); } - return fs; } protected FileSystem createFileSystem(org.apache.hadoop.fs.Path path) throws IOException { Configuration conf = hadoopConf.get(); FileSystem fileSystem = path.getFileSystem(conf); - fileSystem = HadoopSecuredFileSystem.trySecureFileSystem(fileSystem, options, conf); - return fileSystem; + boolean handedOver = false; + try { + // a Kerberos login failure in here would otherwise strand the instance above, which + // with the cache disabled nobody else can reach. This runs only for the creator that + // owns what it builds, so the scheme alone decides, as it does at publication time + FileSystem secured = + HadoopSecuredFileSystem.trySecureFileSystem(fileSystem, options, conf); + handedOver = true; + return secured; + } finally { + if (!handedOver && isOwnedScheme(path.toUri().getScheme())) { + IOUtils.closeQuietly(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. + * + *

Asked while the instance is being created, and the answer is kept next to it in {@link + * #fsMap}. The configuration is shared with the caller and can change underneath us, so asking + * again at close time could contradict what Hadoop actually did. A flip concurrent with the + * creation itself can still be missed; that window is a few instructions rather than the whole + * lifetime of the instance, and nothing short of Hadoop reporting its own decision closes it. + */ + @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, Pair> map = fsMap; + if (map == null) { + return; + } + for (Pair entry : map.values()) { + if (entry.getRight()) { + owned.add(entry.getLeft()); + } + } + // 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/main/java/org/apache/paimon/rest/RESTTokenFileIO.java b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java index fb210dda435f..35cb68eae23e 100644 --- a/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java +++ b/paimon-common/src/main/java/org/apache/paimon/rest/RESTTokenFileIO.java @@ -18,16 +18,21 @@ package org.apache.paimon.rest; +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BlobDescriptor; import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.FileRange; import org.apache.paimon.fs.FileStatus; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.PositionOutputStreamWrapper; import org.apache.paimon.fs.RemoteIterator; import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.SeekableInputStreamWrapper; import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.fs.VectoredReadable; import org.apache.paimon.options.ConfigOption; import org.apache.paimon.options.ConfigOptions; import org.apache.paimon.options.Options; @@ -44,12 +49,18 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + +import java.io.Closeable; import java.io.IOException; import java.io.UncheckedIOException; import java.time.Duration; +import java.util.List; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.paimon.options.CatalogOptions.FILE_IO_ALLOW_CACHE; import static org.apache.paimon.rest.RESTApi.TOKEN_EXPIRATION_SAFE_TIME_MILLIS; @@ -67,12 +78,13 @@ public class RESTTokenFileIO implements FileIO { .defaultValue(false) .withDescription("Whether to support data token provided by the REST server."); - private static final Cache FILE_IO_CACHE = + private static final Cache FILE_IO_CACHE = Caffeine.newBuilder() .maximumSize(1000) .expireAfterAccess(10, TimeUnit.HOURS) - .removalListener( - (ignored, value, cause) -> IOUtils.closeQuietly((FileIO) value)) + // hands back the cache's own reference only, the file system stays alive as + // long as somebody is still reading or writing through it + .removalListener((ignored, value, cause) -> ((CachedFileIO) value).release()) .scheduler( Scheduler.forScheduledExecutorService( Executors.newSingleThreadScheduledExecutor( @@ -109,62 +121,128 @@ public void configure(CatalogContext context) { @Override public SeekableInputStream newInputStream(Path path) throws IOException { - return fileIO().newInputStream(path); + Lease lease = acquire(); + boolean opened = false; + try { + SeekableInputStream delegate = lease.fileIO().newInputStream(path); + // readers pick their strategy with instanceof VectoredReadable, so a wrapper that does + // not carry the capability silently downgrades them to sequential reads + SeekableInputStream in = + delegate instanceof VectoredReadable + ? new LeasedVectoredInputStream(delegate, lease) + : new LeasedSeekableInputStream(delegate, lease); + opened = true; + return in; + } finally { + if (!opened) { + lease.close(); + } + } } @Override public PositionOutputStream newOutputStream(Path path, boolean overwrite) throws IOException { - return fileIO().newOutputStream(path, overwrite); + Lease lease = acquire(); + boolean opened = false; + try { + PositionOutputStream out = + new LeasedPositionOutputStream( + lease.fileIO().newOutputStream(path, overwrite), lease); + opened = true; + return out; + } finally { + if (!opened) { + lease.close(); + } + } } @Override public TwoPhaseOutputStream newTwoPhaseOutputStream(Path path, boolean overwrite) throws IOException { - return fileIO().newTwoPhaseOutputStream(path, overwrite); + Lease lease = acquire(); + boolean opened = false; + try { + TwoPhaseOutputStream out = + new LeasedTwoPhaseOutputStream( + lease.fileIO().newTwoPhaseOutputStream(path, overwrite), lease); + opened = true; + return out; + } finally { + if (!opened) { + lease.close(); + } + } } @Override public FileStatus getFileStatus(Path path) throws IOException { - return fileIO().getFileStatus(path); + try (Lease lease = acquire()) { + return lease.fileIO().getFileStatus(path); + } } @Override public FileStatus[] listStatus(Path path) throws IOException { - return fileIO().listStatus(path); + try (Lease lease = acquire()) { + return lease.fileIO().listStatus(path); + } } @Override public RemoteIterator listFilesIterative(Path path, boolean recursive) throws IOException { // the interface default would hide the inner FileIO's iterative listing override - return fileIO().listFilesIterative(path, recursive); + Lease lease = acquire(); + boolean listing = false; + try { + RemoteIterator iterator = + new LeasedRemoteIterator( + lease.fileIO().listFilesIterative(path, recursive), lease); + listing = true; + return iterator; + } finally { + if (!listing) { + lease.close(); + } + } } @Override public boolean exists(Path path) throws IOException { - return fileIO().exists(path); + try (Lease lease = acquire()) { + return lease.fileIO().exists(path); + } } @Override public boolean delete(Path path, boolean recursive) throws IOException { - return fileIO().delete(path, recursive); + try (Lease lease = acquire()) { + return lease.fileIO().delete(path, recursive); + } } @Override public boolean mkdirs(Path path) throws IOException { - return fileIO().mkdirs(path); + try (Lease lease = acquire()) { + return lease.fileIO().mkdirs(path); + } } @Override public boolean rename(Path src, Path dst) throws IOException { - return fileIO().rename(src, dst); + try (Lease lease = acquire()) { + return lease.fileIO().rename(src, dst); + } } @Override public boolean tryToWriteAtomic(Path path, String content) throws IOException { // the interface default (temp file + rename) would bypass the inner FileIO's atomic // override - return fileIO().tryToWriteAtomic(path, content); + try (Lease lease = acquire()) { + return lease.fileIO().tryToWriteAtomic(path, content); + } } @Override @@ -173,51 +251,111 @@ public String createBlobPresignedUrl( if (!path.equals(tableRoot)) { throw new IOException("Table root does not match RESTTokenFileIO bound table root."); } - return fileIO().createBlobPresignedUrl(tableRoot, descriptor, validity); + try (Lease lease = acquire()) { + return lease.fileIO().createBlobPresignedUrl(tableRoot, descriptor, validity); + } } @Override public boolean isObjectStore() { - try { - return fileIO().isObjectStore(); + try (Lease lease = acquire()) { + return lease.fileIO().isObjectStore(); } catch (IOException e) { throw new RuntimeException(e); } } - public FileIO fileIO() throws IOException { + /** + * The {@link FileIO} for the current token, valid only for as long as the returned {@link + * Lease} is held. Nothing keeps it alive afterwards: the cache is shared by the whole JVM and + * evicting an entry closes the file system behind it. + */ + public Lease acquire() throws IOException { tryToRefreshToken(); - FileIO fileIO = FILE_IO_CACHE.getIfPresent(token); - if (fileIO != null) { - return fileIO; - } - - synchronized (FILE_IO_CACHE) { - fileIO = FILE_IO_CACHE.getIfPresent(token); - if (fileIO != null) { - return fileIO; + while (true) { + RESTToken currentToken = token; + CachedFileIO cached = FILE_IO_CACHE.getIfPresent(currentToken); + if (cached != null) { + Lease lease = cached.acquire(); + if (lease != null) { + return lease; + } + // spent, so it must not be handed out. Caffeine drops the mapping before it + // notifies, which is what keeps this from being reachable today; the branch stays + // because the alternative to it is handing out a closed file system + FILE_IO_CACHE.asMap().remove(currentToken, cached); } - Options options = catalogContext.options(); - options = new Options(RESTUtil.merge(options.toMap(), token.token())); - options.set(FILE_IO_ALLOW_CACHE, false); - CatalogContext context = - CatalogContext.create( - options, - catalogContext.hadoopConf(), - catalogContext.preferIO(), - catalogContext.fallbackIO()); - try { - fileIO = FileIO.get(path, context); - } catch (IOException e) { - throw new UncheckedIOException(e); + synchronized (FILE_IO_CACHE) { + cached = FILE_IO_CACHE.getIfPresent(currentToken); + if (cached != null) { + Lease lease = cached.acquire(); + if (lease != null) { + return lease; + } + FILE_IO_CACHE.asMap().remove(currentToken, cached); + continue; + } + + Options options = catalogContext.options(); + options = new Options(RESTUtil.merge(options.toMap(), currentToken.token())); + options.set(FILE_IO_ALLOW_CACHE, false); + CatalogContext context = + CatalogContext.create( + options, + catalogContext.hadoopConf(), + catalogContext.preferIO(), + catalogContext.fallbackIO()); + FileIO fileIO; + try { + fileIO = FileIO.get(path, context); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + cached = new CachedFileIO(fileIO); + // take the caller's reference before publishing: an admission policy may evict a + // freshly inserted entry straight away, and that must only drop the cache's own + // reference, never close what we are about to hand out + Lease lease = cached.acquire(); + FILE_IO_CACHE.put(currentToken, cached); + return lease; } - FILE_IO_CACHE.put(token, fileIO); - return fileIO; } } + /** + * A {@link Lease} on whatever {@code fileIO} resolves to. Anything other than a {@link + * RESTTokenFileIO} owns its own lifetime, so the lease is a no-op there. + */ + public static Lease lease(FileIO fileIO) throws IOException { + if (fileIO instanceof RESTTokenFileIO) { + return ((RESTTokenFileIO) fileIO).acquire(); + } + return new Lease(fileIO, null); + } + + /** + * The {@link FileIO} for the current token. Its lease is never handed back, because a raw + * reference says nothing about when the caller is done with it, and closing the lease here + * would let an eviction in the same window close the instance on its way out of this method. + * + * @deprecated pins the entry for the life of the process; use {@link #acquire()} and work + * inside the lease. Kept because engines and plugins cast the result to their own concrete + * implementation, which a wrapper would break. + */ + @Deprecated + public FileIO fileIO() throws IOException { + return acquire().fileIO(); + } + + /** Drops every entry. The releases it triggers are dispatched asynchronously by the cache. */ + @VisibleForTesting + static void invalidateFileIOCache() { + FILE_IO_CACHE.invalidateAll(); + FILE_IO_CACHE.cleanUp(); + } + private void tryToRefreshToken() { if (shouldRefresh()) { synchronized (this) { @@ -282,4 +420,256 @@ public RESTToken validToken() { tryToRefreshToken(); return token; } + + /** + * A cached {@link FileIO} and the number of references still outstanding on it. One of them + * belongs to the cache itself and is handed back when the entry is evicted; the rest are leases + * held by callers. The delegate is closed once the last one is gone, so an eviction can never + * shut down a file system that is still serving somebody. + * + *

The flip side is that a lease which is never handed back keeps the file system alive for + * good, and once the entry is out of the cache nothing can reach it to close it. A stream that + * is opened and then abandoned used to cost a stream handle until the next eviction released + * the delegate anyway; now it strands the delegate too. + */ + @VisibleForTesting + static class CachedFileIO { + + private final FileIO fileIO; + private final AtomicInteger references = new AtomicInteger(1); + + CachedFileIO(FileIO fileIO) { + this.fileIO = fileIO; + } + + /** A new lease, or null once the delegate is gone and this instance is unusable. */ + @Nullable + Lease acquire() { + while (true) { + int current = references.get(); + // not just zero: never resurrect on a count that somehow went negative, that would + // hand out a delegate whose close() has already run + if (current <= 0) { + return null; + } + if (references.compareAndSet(current, current + 1)) { + return new Lease(fileIO, this); + } + } + } + + void release() { + if (references.decrementAndGet() == 0) { + IOUtils.closeQuietly(fileIO); + } + } + } + + /** + * The right to use a {@link FileIO} until this lease is closed. Closing it twice releases once. + */ + public static class Lease implements Closeable { + + private final FileIO fileIO; + @Nullable private final CachedFileIO cached; + private final AtomicBoolean released = new AtomicBoolean(); + + private Lease(FileIO fileIO, @Nullable CachedFileIO cached) { + this.fileIO = fileIO; + this.cached = cached; + } + + public FileIO fileIO() { + return fileIO; + } + + @Override + public void close() { + if (cached != null && released.compareAndSet(false, true)) { + cached.release(); + } + } + } + + private static class LeasedSeekableInputStream extends SeekableInputStreamWrapper { + + private final Lease lease; + + private LeasedSeekableInputStream(SeekableInputStream in, Lease lease) { + super(in); + this.lease = lease; + } + + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + lease.close(); + } + } + } + + /** Keeps the vectored read capability of the wrapped stream reachable through the wrapper. */ + private static class LeasedVectoredInputStream extends LeasedSeekableInputStream + implements VectoredReadable { + + private LeasedVectoredInputStream(SeekableInputStream in, Lease lease) { + super(in, lease); + } + + private VectoredReadable vectored() { + return (VectoredReadable) in; + } + + @Override + public int pread(long position, byte[] buffer, int offset, int length) throws IOException { + return vectored().pread(position, buffer, offset, length); + } + + @Override + public void preadFully(long position, byte[] buffer, int offset, int length) + throws IOException { + vectored().preadFully(position, buffer, offset, length); + } + + @Override + public int minSeekForVectorReads() { + return vectored().minSeekForVectorReads(); + } + + @Override + public int batchSizeForVectorReads() { + return vectored().batchSizeForVectorReads(); + } + + @Override + public int parallelismForVectorReads() { + return vectored().parallelismForVectorReads(); + } + + @Override + public void readVectored(List ranges) throws IOException { + vectored().readVectored(ranges); + } + } + + private static class LeasedPositionOutputStream extends PositionOutputStreamWrapper { + + private final Lease lease; + + private LeasedPositionOutputStream(PositionOutputStream out, Lease lease) { + super(out); + this.lease = lease; + } + + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + lease.close(); + } + } + } + + /** + * Both {@link #close()} and {@link #closeForCommit()} end the stream, and implementations call + * one from the other, so both release the lease and rely on it releasing only once. The {@link + * Committer} takes a lease of its own when it runs. + */ + private static class LeasedTwoPhaseOutputStream extends TwoPhaseOutputStream { + + private final TwoPhaseOutputStream out; + private final Lease lease; + + private LeasedTwoPhaseOutputStream(TwoPhaseOutputStream out, Lease lease) { + this.out = out; + this.lease = lease; + } + + @Override + public long getPos() throws IOException { + return out.getPos(); + } + + @Override + public void write(int b) throws IOException { + out.write(b); + } + + @Override + public void write(byte[] b) throws IOException { + out.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + out.write(b, off, len); + } + + @Override + public void flush() throws IOException { + out.flush(); + } + + @Override + public Committer closeForCommit() throws IOException { + try { + return out.closeForCommit(); + } finally { + lease.close(); + } + } + + @Override + public void close() throws IOException { + try { + out.close(); + } finally { + lease.close(); + } + } + } + + /** + * A listing is consumed lazily, so the lease has to outlive the call that started it. {@link + * RemoteIterator} has nothing to close, so the lease goes back once the listing runs dry or + * fails; a caller that walks away mid-listing keeps the entry pinned until the process ends. + */ + private static class LeasedRemoteIterator implements RemoteIterator { + + private final RemoteIterator iterator; + private final Lease lease; + + private LeasedRemoteIterator(RemoteIterator iterator, Lease lease) { + this.iterator = iterator; + this.lease = lease; + } + + @Override + public boolean hasNext() throws IOException { + boolean hasNext; + try { + hasNext = iterator.hasNext(); + } catch (Throwable t) { + lease.close(); + throw t; + } + if (!hasNext) { + lease.close(); + } + return hasNext; + } + + @Override + public FileStatus next() throws IOException { + try { + return iterator.next(); + } catch (Throwable t) { + lease.close(); + throw t; + } + } + } } diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java index 96e023d1eb46..6ceadafa73a6 100644 --- a/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/fs/FileIOTest.java @@ -37,7 +37,9 @@ import java.nio.file.StandardCopyOption; import java.time.Duration; import java.util.Arrays; +import java.util.Collections; import java.util.Comparator; +import java.util.List; import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; @@ -45,6 +47,12 @@ import static org.apache.paimon.utils.Preconditions.checkState; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** Test static methods and methods with default implementations of {@link FileIO}. */ public class FileIOTest { @@ -74,6 +82,61 @@ public void testRequireOptions() throws IOException { assertThat(fileIO).isInstanceOf(RequireOptionsFileIOLoader.MyFileIO.class); } + @Test + public void testCheckedFileIOIsHandedBackInsteadOfReloaded() throws IOException { + // FileIOLoader promises nothing about handing out a fresh instance per call, so a loader + // that keeps one must not get back the instance the access check already released + FileIO singleton = mock(FileIO.class); + when(singleton.exists(any())).thenReturn(true); + + FileIO fileIO = + FileIO.get( + new Path("singleton://bucket/table"), + CatalogContext.create( + new Options(), new SingletonFileIOLoader(singleton, false), null)); + + assertThat(fileIO).isSameAs(singleton); + verify(singleton, never()).close(); + verify(singleton, times(1)).configure(any()); + } + + @Test + public void testCheckedFileIOIsReleasedWhenItsLoaderIsRejected() throws IOException { + // the access check passed, but the loader is out of the running for missing options, so + // the instance it built has to go + FileIO singleton = mock(FileIO.class); + when(singleton.exists(any())).thenReturn(true); + + FileIO fileIO = + FileIO.get( + new Path(tempDir.toUri().toString()), + CatalogContext.create( + new Options(), new SingletonFileIOLoader(singleton, true), null)); + + assertThat(fileIO).isNotSameAs(singleton); + verify(singleton, times(1)).close(); + } + + @Test + public void testCheckedFileIOIsReleasedWhenSelectionBlowsUp() throws IOException { + // FileIOLoader is a public SPI and its methods do throw: FlinkFileIOLoader.getScheme() + // raises UnsupportedOperationException. This drives requiredOptions(), which the selection + // calls unconditionally; whatever it trips over, the checked instance must not be left open + FileIO singleton = mock(FileIO.class); + when(singleton.exists(any())).thenReturn(true); + SingletonFileIOLoader loader = new SingletonFileIOLoader(singleton, false); + loader.failRequiredOptions = true; + + assertThatThrownBy( + () -> + FileIO.get( + new Path("singleton://bucket/table"), + CatalogContext.create(new Options(), loader, null))) + .isInstanceOf(IllegalStateException.class); + + verify(singleton, times(1)).close(); + } + @Test public void testCopy() throws Exception { Path srcFile = new Path(tempDir.resolve("src.txt").toUri()); @@ -341,4 +404,40 @@ private File toFile(Path path) { return new File(localPath); } } + + /** A {@link FileIOLoader} that hands out the same instance every time, as it is free to do. */ + private static class SingletonFileIOLoader implements FileIOLoader { + + private static final long serialVersionUID = 1L; + + private final FileIO fileIO; + private final boolean requireMissingOption; + + private boolean failRequiredOptions; + + private SingletonFileIOLoader(FileIO fileIO, boolean requireMissingOption) { + this.fileIO = fileIO; + this.requireMissingOption = requireMissingOption; + } + + @Override + public String getScheme() { + return "singleton"; + } + + @Override + public List requiredOptions() { + if (failRequiredOptions) { + throw new IllegalStateException("this loader cannot tell"); + } + return requireMissingOption + ? Collections.singletonList(new String[] {"missing-option"}) + : Collections.emptyList(); + } + + @Override + public FileIO load(Path path) { + return fileIO; + } + } } 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 067c7da649aa..e02b28427466 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,17 +29,23 @@ 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.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -146,6 +152,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(); + + // one delegate per authority, each of them released exactly once + assertEquals(2, loaded.size()); + 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, so the test can tell one + * resolved delegate from another. + */ + 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..09071631fe80 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/fs/hadoop/HadoopFileIOTest.java @@ -0,0 +1,557 @@ +/* + * 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.InstantiationUtil; +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 { + // a file system handed in from the outside stays the caller's to close, even where the + // scheme says an instance we built ourselves would have been ours + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "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")); + + try { + assertThat(((RecordingFileSystem) returned).closeCount()).isZero(); + } finally { + ((RecordingFileSystem) returned).close(); + } + } + + @Test + public void testADeserializedCopyOutlivesTheInstanceItCameFrom() throws Exception { + // closing one instance must not disable the copies shipped to other processes, which is + // what the transient closed flag buys + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://serialized/warehouse"); + RecordingFileSystem fs = fileSystem(fileIO, "testfs://serialized/a"); + + byte[] shipped = InstantiationUtil.serializeObject(fileIO); + fileIO.close(); + assertThat(fs.closeCount()).isEqualTo(1); + + HadoopFileIO copy = + InstantiationUtil.deserializeObject(shipped, HadoopFileIO.class.getClassLoader()); + + // the copy starts out usable and owns a file system of its own + RecordingFileSystem copyFs = fileSystem(copy, "testfs://serialized/a"); + assertThat(copyFs).isNotSameAs(fs); + assertThat(copyFs.closeCount()).isZero(); + + copy.close(); + assertThat(copyFs.closeCount()).isEqualTo(1); + } + + @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 testOwnedFileSystemIsStillClosedAfterTheFlagIsFlippedOff() throws Exception { + // the configuration is shared with the caller and can be edited at any time; ownership was + // settled when Hadoop handed out this instance and must not be re-litigated at close time + Configuration conf = conf("testfs"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + HadoopFileIO fileIO = fileIO(conf, "testfs://flippedoff/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://flippedoff/a"); + conf.setBoolean("fs.testfs.impl.disable.cache", false); + + fileIO.close(); + + assertThat(fs.closeCount()).isEqualTo(1); + } + + @Test + public void testSharedFileSystemIsStillSparedAfterTheFlagIsFlippedOn() throws Exception { + // the mirror case: this instance came out of Hadoop's global cache and other readers hold + // it, so a flag flip must not turn it into something we may close + Configuration conf = conf("testfs"); + HadoopFileIO fileIO = fileIO(conf, "testfs://flippedon/warehouse"); + + RecordingFileSystem fs = fileSystem(fileIO, "testfs://flippedon/a"); + conf.setBoolean("fs.testfs.impl.disable.cache", true); + + try { + fileIO.close(); + + assertThat(fs.closeCount()).isZero(); + } finally { + // Hadoop's cache is static and this fork is reused, so never leave it behind + fs.close(); + } + } + + @Test + public void testMalformedDefaultFileSystemDoesNotBreakClose() throws Exception { + // a scheme-less fs.defaultFS must not make ownership blow up. Hadoop 2, which this build + // compiles against, rewrites it to hdfs://; Hadoop 3 throws IllegalArgumentException out of + // getDefaultUri instead, which is what the catch in isOwnedScheme is there for + 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 testAccessProbeFileSystemIsReusedAndNotLeakedByFileIOGet() throws Exception { + // FileIO.get probes a loader by calling exists() on a FileIO it then hands back. With the + // Hadoop cache disabled a discarded probe would strand a file system nobody can reach, and + // reloading instead of reusing would build a second one for no reason + 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); + assertThat(RecordingFileSystem.created()).isEqualTo(1); + } finally { + fileIO.close(); + } + + 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()), + Pair.of(winner == null ? loser : winner, isOwnedScheme(uri.getScheme()))); + 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)); diff --git a/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java b/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java index 9bab3e1a8976..ada42cb180b4 100644 --- a/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/rest/RESTTokenFileIOTest.java @@ -21,32 +21,63 @@ import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.fs.BaseMultiPartUploadCommitter; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.FileIOLoader; import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.MultiPartUploadStore; import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.RemoteIterator; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.fs.VectoredReadable; import org.apache.paimon.options.Options; import org.apache.paimon.rest.responses.GetTableTokenResponse; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.Mockito; import java.io.IOException; import java.time.Duration; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; import java.util.UUID; +import java.util.concurrent.atomic.AtomicBoolean; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.after; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; /** Tests for {@link RESTTokenFileIO}. */ class RESTTokenFileIOTest { + private static final Path TABLE_ROOT = new Path("oss://bucket/table"); + private static final Path FILE = new Path("oss://bucket/table/bucket-0/data"); + + /** The cache hands releases to an executor, so a close lands shortly after, not inline. */ + private static final long CLOSED_MILLIS = 30_000; + + private static final long NOT_CLOSED_MILLIS = 500; + + @BeforeEach + @AfterEach + void clearFileIOCache() { + // the cache is static and shared by every test in this fork + RESTTokenFileIO.invalidateFileIOCache(); + } + @Test void testCreateBlobPresignedUrlRequiresBoundRootAndDelegates() throws IOException { Path tableRoot = new Path("oss://bucket/table"); @@ -164,4 +195,307 @@ public FileStatus next() { // the interface default would construct its own iterator backed by listStatus verify(delegate, never()).listStatus(any()); } + + @Test + void testEvictionKeepsALeasedFileIOAlive() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("leased", delegates); + + RESTTokenFileIO.Lease lease = fileIO.acquire(); + FileIO delegate = delegates.get(0); + assertThat(lease.fileIO()).isSameAs(delegate); + + RESTTokenFileIO.invalidateFileIOCache(); + + // the cache handed its own reference back, ours is still out + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + assertThat(lease.fileIO().exists(FILE)).isTrue(); + + lease.close(); + + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testEvictionClosesAnUnleasedFileIO() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("unleased", delegates); + + fileIO.exists(FILE); + + RESTTokenFileIO.invalidateFileIOCache(); + + verify(delegates.get(0), timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testAnOpenStreamKeepsTheFileIOAlive() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("stream", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + SeekableInputStream delegateStream = mock(SeekableInputStream.class); + when(delegate.newInputStream(FILE)).thenReturn(delegateStream); + + SeekableInputStream in = fileIO.newInputStream(FILE); + RESTTokenFileIO.invalidateFileIOCache(); + + // a read in flight outlives the eviction that dropped its cache entry + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + + in.close(); + + verify(delegateStream, times(1)).close(); + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testAnOutputStreamKeepsTheFileIOAlive() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("output", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + PositionOutputStream delegateStream = mock(PositionOutputStream.class); + when(delegate.newOutputStream(FILE, false)).thenReturn(delegateStream); + + PositionOutputStream out = fileIO.newOutputStream(FILE, false); + RESTTokenFileIO.invalidateFileIOCache(); + + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + + out.close(); + + verify(delegateStream, times(1)).close(); + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testTheLeasedStreamKeepsTheVectoredReadCapability() throws Exception { + // readers choose their strategy with instanceof VectoredReadable, so a wrapper that hides + // it would silently downgrade every REST catalog read to sequential + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("vectored", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + VectoredSeekableInputStream delegateStream = mock(VectoredSeekableInputStream.class); + when(delegate.newInputStream(FILE)).thenReturn(delegateStream); + when(delegateStream.pread(1L, new byte[0], 2, 3)).thenReturn(7); + when(delegateStream.parallelismForVectorReads()).thenReturn(11); + + SeekableInputStream in = fileIO.newInputStream(FILE); + assertThat(in).isInstanceOf(VectoredReadable.class); + VectoredReadable vectored = (VectoredReadable) in; + assertThat(vectored.pread(1L, new byte[0], 2, 3)).isEqualTo(7); + assertThat(vectored.parallelismForVectorReads()).isEqualTo(11); + + // the capability-carrying wrapper still has to hand its lease back + RESTTokenFileIO.invalidateFileIOCache(); + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + in.close(); + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testAPlainStreamIsNotDressedUpAsVectored() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("plainstream", delegates); + fileIO.exists(FILE); + when(delegates.get(0).newInputStream(FILE)).thenReturn(mock(SeekableInputStream.class)); + + try (SeekableInputStream in = fileIO.newInputStream(FILE)) { + assertThat(in).isNotInstanceOf(VectoredReadable.class); + } + } + + @Test + void testATwoPhaseStreamReleasesItsLeaseOnCloseForCommit() throws Exception { + // the load-bearing one: FormatTableSingleFileWriter only ever calls closeForCommit() + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("twophasecommit", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + when(delegate.newTwoPhaseOutputStream(FILE, false)) + .thenReturn(mock(TwoPhaseOutputStream.class)); + + TwoPhaseOutputStream out = fileIO.newTwoPhaseOutputStream(FILE, false); + RESTTokenFileIO.invalidateFileIOCache(); + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + + out.closeForCommit(); + + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testATwoPhaseStreamReleasesItsLeaseOnClose() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("twophaseclose", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + when(delegate.newTwoPhaseOutputStream(FILE, false)) + .thenReturn(mock(TwoPhaseOutputStream.class)); + + TwoPhaseOutputStream out = fileIO.newTwoPhaseOutputStream(FILE, false); + RESTTokenFileIO.invalidateFileIOCache(); + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + + out.close(); + + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testAFailedOpenHandsTheLeaseBack() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("failedopen", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + when(delegate.newInputStream(FILE)).thenThrow(new IOException("cannot open")); + + assertThatThrownBy(() -> fileIO.newInputStream(FILE)).isInstanceOf(IOException.class); + + // a lease stranded by the failure would keep the eviction from ever closing the delegate + RESTTokenFileIO.invalidateFileIOCache(); + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testAClosedFileIOIsRebuiltRatherThanHandedOut() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("rebuild", delegates); + + fileIO.exists(FILE); + RESTTokenFileIO.invalidateFileIOCache(); + verify(delegates.get(0), timeout(CLOSED_MILLIS).times(1)).close(); + + fileIO.exists(FILE); + + assertThat(delegates).hasSize(2); + verify(delegates.get(1), never()).close(); + } + + @Test + void testClosingALeaseTwiceReleasesOnce() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("idempotent", delegates); + + RESTTokenFileIO.Lease lease = fileIO.acquire(); + lease.close(); + lease.close(); + + // nothing was evicted yet, so an over-release is the only way this could have closed + verify(delegates.get(0), after(NOT_CLOSED_MILLIS).never()).close(); + + RESTTokenFileIO.invalidateFileIOCache(); + verify(delegates.get(0), timeout(CLOSED_MILLIS).times(1)).close(); + } + + @Test + void testHolderRefusesToHandOutAReleasedFileIO() throws IOException { + FileIO delegate = mock(FileIO.class); + RESTTokenFileIO.CachedFileIO cached = new RESTTokenFileIO.CachedFileIO(delegate); + + RESTTokenFileIO.Lease lease = cached.acquire(); + assertThat(lease).isNotNull(); + + // the reference the cache owns, handed back the way an eviction does + cached.release(); + verify(delegate, after(NOT_CLOSED_MILLIS).never()).close(); + + lease.close(); + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + + // spent: a lookup that raced with the eviction has to be told to build a new one + assertThat(cached.acquire()).isNull(); + } + + @Test + void testLeaseOnAPlainFileIODoesNothing() throws IOException { + FileIO plain = mock(FileIO.class); + + try (RESTTokenFileIO.Lease lease = RESTTokenFileIO.lease(plain)) { + assertThat(lease.fileIO()).isSameAs(plain); + } + + verify(plain, never()).close(); + } + + @Test + void testMultiPartUploadCommitterHoldsALeaseAcrossTheCall() throws Exception { + List delegates = new ArrayList<>(); + RESTTokenFileIO fileIO = fileIO("committer", delegates); + fileIO.exists(FILE); + FileIO delegate = delegates.get(0); + + @SuppressWarnings("unchecked") + MultiPartUploadStore store = mock(MultiPartUploadStore.class); + AtomicBoolean closedDuringUpload = new AtomicBoolean(); + when(store.completeMultipartUpload(any(), any(), any(), anyLong())) + .thenAnswer( + ignored -> { + // the entry goes away while the upload is being completed + RESTTokenFileIO.invalidateFileIOCache(); + Thread.sleep(NOT_CLOSED_MILLIS); + closedDuringUpload.set( + Mockito.mockingDetails(delegate).getInvocations().stream() + .anyMatch( + i -> "close".equals(i.getMethod().getName()))); + return "done"; + }); + + TestCommitter committer = new TestCommitter<>(store, FILE); + committer.commit(fileIO); + + verify(store, times(1)).completeMultipartUpload(any(), any(), any(), anyLong()); + assertThat(closedDuringUpload).isFalse(); + verify(delegate, timeout(CLOSED_MILLIS).times(1)).close(); + // the store must be built on the delegate, plugin committers cast it to their own impl + assertThat(committer.received).isSameAs(delegate); + } + + private RESTTokenFileIO fileIO(String tokenValue, List delegates) { + 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); + delegates.add(delegate); + return delegate; + }); + RESTApi api = mock(RESTApi.class); + Identifier identifier = Identifier.create("db", "table"); + // a token of its own per test, so the shared cache keeps the entries apart + when(api.loadTableToken(identifier)) + .thenReturn( + new GetTableTokenResponse( + Collections.singletonMap("token", tokenValue), Long.MAX_VALUE)); + return new RESTTokenFileIO( + CatalogContext.create(new Options(), loader, null), api, identifier, TABLE_ROOT); + } + + /** The shape of a delegate stream whose capability the leased wrapper has to carry through. */ + private abstract static class VectoredSeekableInputStream extends SeekableInputStream + implements VectoredReadable {} + + private static class TestCommitter extends BaseMultiPartUploadCommitter { + + private static final long serialVersionUID = 1L; + + private final transient MultiPartUploadStore store; + + private transient FileIO received; + + private TestCommitter(MultiPartUploadStore store, Path targetPath) { + super("upload-id", Collections.emptyList(), "object", 0L, targetPath); + this.store = store; + } + + @Override + protected MultiPartUploadStore multiPartUploadStore(FileIO fileIO, Path targetPath) { + received = fileIO; + return store; + } + } } diff --git a/paimon-lance/src/main/java/org/apache/paimon/format/lance/LanceUtils.java b/paimon-lance/src/main/java/org/apache/paimon/format/lance/LanceUtils.java index 9d5828a16602..005c02585621 100644 --- a/paimon-lance/src/main/java/org/apache/paimon/format/lance/LanceUtils.java +++ b/paimon-lance/src/main/java/org/apache/paimon/format/lance/LanceUtils.java @@ -67,21 +67,7 @@ private static Pair> toLanceSpecified( URI uri = path.toUri(); String schema = uri.getScheme(); - if (fileIO instanceof RESTTokenFileIO) { - try { - fileIO = ((RESTTokenFileIO) fileIO).fileIO(); - } catch (IOException e) { - throw new RuntimeException("Can't get fileIO from RESTTokenFileIO", e); - } - } - - Options originOptions; - if (fileIO instanceof HadoopOptionsProvider) { - originOptions = - ((HadoopOptionsProvider) fileIO).hadoopOptions(path, isRead ? "read" : "write"); - } else { - originOptions = new Options(); - } + Options originOptions = hadoopOptions(fileIO, path, isRead); Path converted = path; Map storageOptions = new HashMap<>(); @@ -128,4 +114,27 @@ private static Pair> toLanceSpecified( return Pair.of(converted, storageOptions); } + + /** + * The options of the file system behind {@code fileIO}. A REST token FileIO resolves to an + * instance out of a cache shared by the JVM, so the read happens under a lease: without one an + * eviction can close that instance while it is being asked. + */ + private static Options hadoopOptions(FileIO fileIO, Path path, boolean isRead) { + String opType = isRead ? "read" : "write"; + if (fileIO instanceof RESTTokenFileIO) { + try (RESTTokenFileIO.Lease lease = ((RESTTokenFileIO) fileIO).acquire()) { + return hadoopOptions(lease.fileIO(), path, opType); + } catch (IOException e) { + throw new RuntimeException("Can't get fileIO from RESTTokenFileIO", e); + } + } + return hadoopOptions(fileIO, path, opType); + } + + private static Options hadoopOptions(FileIO fileIO, Path path, String opType) { + return fileIO instanceof HadoopOptionsProvider + ? ((HadoopOptionsProvider) fileIO).hadoopOptions(path, opType) + : new Options(); + } } diff --git a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java index c704ed73cb34..f10aaa17d93f 100644 --- a/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java +++ b/paimon-vortex/paimon-vortex-format/src/main/java/org/apache/paimon/format/vortex/VortexUtils.java @@ -48,21 +48,7 @@ private static Pair> toVortexSpecified( URI uri = path.toUri(); String schema = uri.getScheme(); - if (fileIO instanceof RESTTokenFileIO) { - try { - fileIO = ((RESTTokenFileIO) fileIO).fileIO(); - } catch (IOException e) { - throw new RuntimeException("Can't get fileIO from RESTTokenFileIO", e); - } - } - - Options originOptions; - if (fileIO instanceof HadoopOptionsProvider) { - originOptions = - ((HadoopOptionsProvider) fileIO).hadoopOptions(path, isRead ? "read" : "write"); - } else { - originOptions = new Options(); - } + Options originOptions = hadoopOptions(fileIO, path, isRead); Path converted = path; Map storageOptions = new HashMap<>(); @@ -81,4 +67,27 @@ private static Pair> toVortexSpecified( return Pair.of(converted, storageOptions); } + + /** + * The options of the file system behind {@code fileIO}. A REST token FileIO resolves to an + * instance out of a cache shared by the JVM, so the read happens under a lease: without one an + * eviction can close that instance while it is being asked. + */ + private static Options hadoopOptions(FileIO fileIO, Path path, boolean isRead) { + String opType = isRead ? "read" : "write"; + if (fileIO instanceof RESTTokenFileIO) { + try (RESTTokenFileIO.Lease lease = ((RESTTokenFileIO) fileIO).acquire()) { + return hadoopOptions(lease.fileIO(), path, opType); + } catch (IOException e) { + throw new RuntimeException("Can't get fileIO from RESTTokenFileIO", e); + } + } + return hadoopOptions(fileIO, path, opType); + } + + private static Options hadoopOptions(FileIO fileIO, Path path, String opType) { + return fileIO instanceof HadoopOptionsProvider + ? ((HadoopOptionsProvider) fileIO).hadoopOptions(path, opType) + : new Options(); + } }