wombatu-kun commented on code in PR #8962:
URL: https://github.com/apache/paimon/pull/8962#discussion_r3705148648
##########
paimon-common/src/main/java/org/apache/paimon/fs/PluginFileIO.java:
##########
@@ -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();
Review Comment:
Done 57edbe84d. Cache values are reference counted now: the removal listener
hands back only the cache's own reference, and the delegate is closed when the
last lease goes, with leases held for the duration of each operation and for
the lifetime of every returned stream.
The window is wider than the entry count suggests - the admission policy can
evict a just-inserted entry, so the caller's lease is taken before the put.
`fileIO()` is deprecated because a raw reference carries no lifetime to track;
`BaseMultiPartUploadCommitter`, `LanceUtils` and `VortexUtils` now work inside
a lease instead.
##########
paimon-common/src/main/java/org/apache/paimon/fs/hadoop/HadoopFileIO.java:
##########
@@ -226,6 +259,64 @@ protected FileSystem
createFileSystem(org.apache.hadoop.fs.Path path) throws IOE
return fileSystem;
}
+ /**
+ * Whether the {@link FileSystem} instances created for the given scheme
belong to this {@link
+ * FileIO} exclusively, and may therefore be closed by it.
+ *
+ * <p>This mirrors the branch Hadoop itself takes in {@code
FileSystem#get(URI, Configuration)}:
+ * with {@code fs.<scheme>.impl.disable.cache} set, Hadoop hands out a
fresh instance that
+ * nobody else can reach, so releasing it is our responsibility. Otherwise
the instance lives in
+ * Hadoop's global cache and is shared with every other user in this JVM,
including other {@link
+ * FileIO}s and the compute engine itself; {@code FileSystem#closeAll}
releases those on
+ * shutdown and closing one here would break unrelated readers.
+ *
+ * <p>The scheme is the one taken from the path, not from {@code
FileSystem#getUri()}, and it is
+ * matched as written rather than lower cased, because that is what Hadoop
looks up. Any
+ * deviation could report a cached, shared instance as owned.
+ */
+ @VisibleForTesting
+ boolean isOwnedScheme(@Nullable String scheme) {
+ if (hadoopConf == null) {
+ return false;
+ }
+ Configuration conf = hadoopConf.get();
+ if (scheme == null) {
+ // a path without a scheme is served by the default file system
+ try {
+ scheme = FileSystem.getDefaultUri(conf).getScheme();
+ } catch (IllegalArgumentException e) {
+ // a missing or malformed fs.defaultFS, so there is no scheme
to claim ownership of
+ return false;
+ }
+ }
+ return conf.getBoolean(String.format("fs.%s.impl.disable.cache",
scheme), false);
+ }
+
+ @Override
+ public void close() throws IOException {
+ List<FileSystem> owned = new ArrayList<>();
+ synchronized (this) {
+ closed = true;
+ Map<Pair<String, String>, FileSystem> map = fsMap;
+ if (map == null) {
+ return;
+ }
+ for (Map.Entry<Pair<String, String>, FileSystem> entry :
map.entrySet()) {
+ if (isOwnedScheme(entry.getKey().getLeft())) {
Review Comment:
Done 57edbe84d. The map holds `{fileSystem, ownedAtCreation}` and both the
loser cleanup and `close()` read that bit. The same pass stops recording an
externally injected file system as owned, releases the raw instance when the
Kerberos wrapper fails, and moves the loser's close outside the monitor so a
slow object store teardown cannot stall the other callers.
##########
paimon-common/src/main/java/org/apache/paimon/fs/FileIO.java:
##########
@@ -655,10 +660,16 @@ static FileIOLoader checkAccess(FileIOLoader fileIO, Path
path, CatalogContext c
return null;
}
- // check access
+ // check access, the probe is thrown away afterwards so it has to be
released here: with
+ // the Hadoop file system cache disabled its exists() call creates a
file system that no
+ // one else can reach
FileIO io = fileIO.load(path);
- io.configure(config);
- io.exists(path);
+ try {
+ io.configure(config);
+ io.exists(path);
+ } finally {
+ IOUtils.closeQuietly(io);
Review Comment:
Done 57edbe84d. `checkAccess` returns the instance it checked and
`FileIO.get` hands that one back, which also stops the file system being built
twice. That changed its return type from `FileIOLoader` to `FileIO`, a
signature change on a `@Public` interface with no other callers in the repo.
The selection is also wrapped now, because an unchecked failure from a loader's
`requiredOptions()` or `getScheme()` would otherwise strand the checked
instance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
