JingsongLi commented on code in PR #8962:
URL: https://github.com/apache/paimon/pull/8962#discussion_r3703449625
##########
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:
**[P2] Preserve the ownership decision made when the filesystem was created**
`FileSystem.get` decides cached-vs-owned using
`fs.<scheme>.impl.disable.cache` at creation time, but this code recomputes
ownership from the current mutable `Configuration`. `SerializableConfiguration`
retains the caller's configuration by reference, `hadoopConf()` exposes it, and
`configure()` can replace it. If a shared cached filesystem is created with the
flag false and the flag later becomes true, `close()` misclassifies and closes
a JVM-global filesystem still used by other readers; the reverse transition
leaks an owned filesystem. Please store `{fileSystem, ownedAtCreation}`
atomically in the map and use that recorded bit for loser cleanup and final
close.
##########
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:
**[P1] Keep REST-token cache entries alive while they are still borrowed**
This forwarding close makes `RESTTokenFileIO`'s existing removal listener
actually close the plugin delegate. That cache returns raw `FileIO` values and
streams without a lease or reference count, so size eviction after more than
1,000 token entries (or expiry) can close an uncached OSS filesystem while
another table is still reading or writing through it.
`AliyunOSSFileSystem.close()` shuts down the OSS client and executor pools, so
the active operation can fail mid-flight. Please defer the physical close until
active operations and returned streams release their leases, or otherwise add
equivalent lifetime tracking around cached values.
##########
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:
**[P2] Do not assume a public `FileIOLoader` returns a fresh instance on
every call**
`checkAccess` now terminally closes the probed instance, but after this
method returns `FileIO.get` calls `loader.load(path)` again. The `@Public
FileIOLoader.load(Path)` contract does not require a fresh instance, so a valid
prefer/fallback loader that caches or returns a singleton will return the same
now-closed `PluginFileIO`, `HadoopFileIO`, or `ResolvingFileIO`; every
subsequent operation then fails with `This FileIO is closed.` Please carry the
successfully probed instance through selection and return it, closing it only
if that candidate is rejected.
--
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]
