Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wombatu-kun wants to merge 2 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1757cf9
[common] Fix file system leaks when the Hadoop file system cache is d…
Jul 31, 2026
57edbe8
[common] Address review comments on file system lifetime
Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 9 additions & 12 deletions paimon-common/src/main/java/org/apache/paimon/fs/BaseMultiPartUploadCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,12 @@ protected abstract MultiPartUploadStore<T, C> multiPartUploadStore(

@Override
public void commit(FileIO fileIO) throws IOException {
try {
MultiPartUploadStore<T, C> 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<T, C> multiPartUploadStore =
multiPartUploadStore(lease.fileIO(), targetPath());
multiPartUploadStore.completeMultipartUpload(
objectName, uploadId, uploadedParts, byteLength);
} catch (Exception e) {
Expand All @@ -70,8 +74,9 @@ public void commit(FileIO fileIO) throws IOException {

@Override
public void discard(FileIO fileIO) throws IOException {
try {
MultiPartUploadStore<T, C> multiPartUploadStore = multiPartUploadStore(fileIO);
try (RESTTokenFileIO.Lease lease = RESTTokenFileIO.lease(fileIO)) {
MultiPartUploadStore<T, C> multiPartUploadStore =
multiPartUploadStore(lease.fileIO(), targetPath());
multiPartUploadStore.abortMultipartUpload(objectName, uploadId);
} catch (Exception e) {
LOG.warn("Failed to discard multipart upload with ID: {}", uploadId, e);
Expand All @@ -90,12 +95,4 @@ public List<T> uploadedParts() {

@Override
public void clean(FileIO fileIO) throws IOException {}

private MultiPartUploadStore<T, C> multiPartUploadStore(FileIO fileIO) throws IOException {
if (fileIO instanceof RESTTokenFileIO) {
RESTTokenFileIO restTokenFileIO = (RESTTokenFileIO) fileIO;
fileIO = restTokenFileIO.fileIO();
}
return multiPartUploadStore(fileIO, targetPath());
}
}
279 changes: 167 additions & 112 deletions paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -270,8 +270,13 @@ default String createBlobPresignedUrl(
}

/**
* Override this method to empty, many FileIO implementation classes rely on static variables
* and do not have the ability to close them.
* Releases the resources this instance owns exclusively. The default is empty because many
* implementations hold nothing of their own, or reach their resources through static variables
* shared with the rest of the JVM, which they must not close.
*
* <p>Override it only for resources that belong to this instance alone, and make the override
* idempotent. Implementations that delegate to another {@link FileIO} should forward the call,
* otherwise the delegate can never be released.
*/
@Override
default void close() throws IOException {}
Expand Down Expand Up @@ -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);
}
}

Expand Down Expand Up @@ -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<IOException> 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<String, FileIOLoader> 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<String, FileIOLoader> 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<String> options =
config.options().keySet().stream()
.map(String::toLowerCase)
.collect(Collectors.toSet());
Set<String> 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<String> options =
config.options().keySet().stream()
.map(String::toLowerCase)
.collect(Collectors.toSet());
Set<String> 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. */
Expand All @@ -649,16 +686,34 @@ static Map<String, FileIOLoader> 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.
*
* <p>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;
}
}
Loading
Loading