whua3 opened a new issue, #11303:
URL: https://github.com/apache/gravitino/issues/11303
### Version
main branch
### Describe what's wrong
### Title
[Bug][gvfs] FileSystem instance can be closed while in use, causing
RejectedExecutionException on long-running writes (Spark batch, etc.)
### Component(s)
clients/filesystem-hadoop3 (GravitinoVirtualFileSystem)
### Affected versions
0.7.x ~ main
### Describe the bug
`BaseGVFSOperations.internalFileSystemCache` is a Caffeine cache of
`(filesetIdent, locationName) -> FileSystem` configured with:
- `expireAfterAccess(evictionMillsAfterAccess)` (default **1 hour**)
- a `removalListener` that calls **`fileSystem.close()`** on eviction
- a dedicated `ScheduledThreadPoolExecutor("gvfs-filesystem-cache-cleaner")`
to proactively trigger eviction even without traffic
```java
// BaseGVFSOperations.java#newFileSystemCache
.removalListener((key, value, cause) -> {
FileSystem fs = (FileSystem) value;
if (fs != null) {
try { fs.close(); } catch (IOException e) { ... }
}
})
.expireAfterAccess(evictionMillsAfterAccess, TimeUnit.MILLISECONDS);
```
The cache's "access" is **only** updated when callers go through
`internalFileSystemCache.get(...)` (i.e.
`getActualFileSystemByLocationName`).
Once a caller has obtained a `FileSystem` handle and continues to use it
through long-lived streams (e.g. `FSDataOutputStream` on object stores),
**no further `cache.get(...)` happens**, so the entry's last-access time
is frozen at stream-open time.
If the stream remains open longer than `evictionMillsAfterAccess`, the
cache evicts the entry and the `removalListener` calls `fs.close()` while
the user code still holds a live reference and is actively writing to it.
For object-store implementations this shuts down internal IO thread pools,
so the next async upload submitted by the open stream throws
`RejectedExecutionException`.
### Reproduce
A Spark job that writes Parquet to a `gvfs://` path backed by `cosn`/`s3a`
and whose per-task duration exceeds 1 hour (default eviction).
After ~1h all in-flight write tasks fail with:
```
java.util.concurrent.RejectedExecutionException:
The bounded io thread pool has been shutdown
at
org.apache.hadoop.fs.CosNFileSystem$1.rejectedExecution(CosNFileSystem.java:209)
at
org.apache.hadoop.fs.CosNFSDataOutputStream$MultipartUpload.uploadPartAsync(...)
at org.apache.hadoop.fs.CosNFSDataOutputStream.uploadCurrentPart(...)
at org.apache.hadoop.fs.CosNFSDataOutputStream.write(...)
at org.apache.parquet.hadoop.ParquetFileWriter.writeColumnChunk(...)
...
Suppressed: java.io.IOException:
The file being written is in an invalid state. ... Current state: COLUMN
```
The same problem affects any long-running reader/writer:
- Spark Structured Streaming sinks that hold an output stream across batches
- Hive/Trino INSERT producing many large files in a single task
- Flink stateful sinks
- Any user code that obtains `FileSystem` once and uses it for a long time
### Why this is a design issue, not just a config tuning issue
Bumping `fs.gravitino.fileset.cache.evictionMillsAfterAccess` or
`maxCapacity` only **reduces the probability** of hitting this. The root
problem is that **eviction is not synchronized with active usage**: the
cache has no idea whether anyone still holds a reference to the cached
`FileSystem`. As long as the cache feels free to call `close()` while a
user holds the handle, the bug exists.
This is also inconsistent with how Hadoop's own `FileSystem.CACHE`
behaves: Hadoop never proactively closes a cached `FileSystem` based on
time; it only closes on `FileSystem.closeAll()` / `closeAllForUGI` /
explicit `close()`. GVFS introduced an `expireAfterAccess +
removalListener.close()` policy that has no analogue in Hadoop and is
the source of this race.
### Prior art
This exact race was first reported in **#3928** (June 2024) with the
same symptom ("Filesystem to be closed while the stream is still working").
It was "resolved" by **#3932**, which only increased the default
`evictionMillsAfterAccess` from 5 minutes to 1 hour - a 3-line change
that does not fix the underlying race, only enlarges the window. Any
long-running task whose per-task duration exceeds the new default still
hits the same bug; we reproduced it on Spark batch jobs writing Parquet
to `cosn://` in May 2026 (single task duration > 1h).
For reference, the **same anti-pattern was later replicated server-side**
in **#7782** (`FilesetCatalogOperations` introduced its own
`Caffeine.expireAfterAccess(1h) + removalListener.close()` cache,
August 2025). That instance is out of scope for this PR but should be
fixed in a follow-up using the same approach.
### Proposed fix
Bring GVFS back in line with Hadoop's own `FileSystem.CACHE` semantics:
**a cached `FileSystem` is never proactively closed by the cache.** Its
lifecycle is bound to the owning `BaseGVFSOperations` instance, not to
the cache's access pattern.
Concretely:
1. Remove the `fs.close()` call from the cache's `removalListener`.
Eviction (size or time based) only detaches the entry from the cache
map; the underlying `FileSystem` continues to be usable by anyone
still holding a reference to it.
2. Maintain a separate set of all `FileSystem` instances ever created
by this operations instance, so that `BaseGVFSOperations#close()`
can close them all at the unambiguous end of the GVFS lifecycle
(Spark App shutdown, explicit close, etc.) -- including those that
have already been evicted from the cache.
3. The cache continues to provide its primary value: avoiding the cost
of repeatedly creating expensive `FileSystem` instances for the same
`(filesetIdent, locationName)` key. It just no longer doubles as a
resource-lifecycle controller.
This is a **~30-line change**, fully backward compatible, and follows
the same model that Hadoop has used successfully for over a decade.
### Backward compatibility
- Default `evictionMillsAfterAccess` and `maxCapacity` values unchanged.
- Cache hit/miss semantics unchanged; existing tests pass without
modification (the cache still returns the same `FileSystem` for the
same key as before).
- The only observable change: an evicted entry's underlying `FileSystem`
is **not** closed at eviction time; instead it is closed when the
owning `BaseGVFSOperations` is closed.
### Trade-offs (called out explicitly for reviewers)
- The cache becomes a "fast lookup index" rather than a "hard upper
bound on resources". For long-running GVFS instances that access a
very large number of distinct `(filesetIdent, locationName)` tuples,
evicted-but-not-yet-closed `FileSystem`s accumulate until shutdown.
- **In practice this is a non-issue**: a single Spark/Flink/Hive
application typically reads/writes from O(1)-O(10) filesets, well
below the default `maxCapacity=20`. Operators with extreme cardinality
can size `maxCapacity` accordingly; they will simply observe more
resident FS instances in exchange for stream-safety.
- This trade-off is exactly the one Hadoop made in its own
`FileSystem.CACHE` — and that hasn't been a problem in 15 years.
### Alternatives considered
1. **Just increase the default eviction time to 24h.** Mitigates but
does not fix; any task longer than the new value still breaks.
**This is exactly what #3932 did in 2024 (5min → 1h) and it did not
prevent the regression we are reporting here.**
2. **Disable eviction altogether.** Equivalent to setting
`evictionMillsAfterAccess` to `Long.MAX_VALUE`; users can already do
this today, but it is opt-in and most users hit the bug before
discovering they need to.
3. **Reference counting (`RefCountedFileSystem` + acquire/release).**
Solves the bug but introduces ~250 lines of new code, requires
wrapping every returned `FSDataOutputStream`/`FSDataInputStream`
(risk of breaking `Syncable`/`StreamCapabilities` interfaces),
and silently leaks if any caller forgets to close a stream. Overkill
for the actual problem; the lifecycle-bound approach in this PR is
cleaner and matches Hadoop's own choice.
4. **Lock during operations.** Doesn't help, because the user holds the
`FSDataOutputStream` long after the operation method returns.
### Willingness to contribute
I'd like to implement this. Will follow up with a PR.
### Error message and/or stacktrace
When `BaseGVFSOperations` is shared across threads, an `IOException:
Filesystem closed` (HDFS) /
`RejectedExecutionException` (object stores) is thrown intermittently during
normal read/write
calls. Example stack:
org.apache.spark.SparkException: [TASK_WRITE_FAILED] Task failed while
writing rows to gvfs://fileset/xxxxxx/xxxxxx/xxxxxxx/20260423.
at
org.apache.spark.sql.errtingQueryExecutionErrors$.taskFailedWhileWritingRowsError(QueryExecutionErrors.scala:775)
at
org.apache.spark.sql.execution.datasources.FileFormatWriter$.executeTask(FileFormatWriter.scala:469)
at
org.apache.spark.sql.execution.datasources.WriteFilesExec.$anonfun$doExecuteWrite$1(WriteFiles.scala:100)
at
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:893)
at
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:893)
at
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:367)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:331)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:94)
at
org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:166)
at org.apache.spark.scheduler.Task.run(Task.scala:146)
at
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:628)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:64)
at
org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:61)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:94)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:631)
at
java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at
java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.util.concurrent.RejectedExecutionException: The bounded io
thread pool has been shutdown
at
org.apache.hadoop.fs.CosNFileSystem$1.rejectedExecution(CosNFileSystem.java:209)
at
java.base/java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:825)
at
java.base/java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1355)
at
com.google.common.util.concurrent.MoreExecutors$ListeningDecorator.execute(MoreExecutors.java:484)
at
java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:140)
at
com.google.common.util.concurrent.AbstractListeningExecutorService.submit(AbstractListeningExecutorService.java:58)
at
org.apache.hadoop.fs.CosNFSDataOutputStream$MultipartUpload.uploadPartAsync(CosNFSDataOutputStream.java:560)
at
org.apache.hadoop.fs.CosNFSDataOutputStream.uploadCurrentPart(CosNFSDataOutputStream.java:435)
at
org.apache.hadoop.fs.CosNFSDataOutputStream.write(CosNFSDataOutputStream.java:163)
at
org.apache.hadoop.fs.FSDataOutputStream$PositionCache.write(FSDataOutputStream.java:62)
at java.base/java.io.DataOutputStream.write(DataOutputStream.java:107)
at
java.base/java.io.FilterOutputStream.write(FilterOutputStream.java:108)
at
org.apache.parquet.hadoop.util.HadoopPositionOutputStream.write(HadoopPositionOutputStream.java:45)
at
org.apache.parquet.bytes.ConcatenatingByteArrayCollector.writeAllTo(ConcatenatingByteArrayCollector.java:46)
at
org.apache.parquet.hadoop.ParquetFileWriter.writeColumnChunk(ParquetFileWriter.java:903)
at
org.apache.parquet.hadoop.ParquetFileWriter.writeColumnChunk(ParquetFileWriter.java:848)
at
org.apache.parquet.hadoop.ColumnChunkPageWriteStore$ColumnChunkPageWriter.writeToFileWriter(ColumnChunkPageWriteStore.java:310)
at
org.apache.parquet.hadoop.ColumnChunkPageWriteStore.flushToFileWriter(ColumnChunkPageWriteStore.java:458)
at
org.apache.parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:186)
at
org.apache.parquet.hadoop.InternalParquetRecordWriter.close(InternalParquetRecordWriter.java:124)
at
org.apache.parquet.hadoop.ParquetRecordWriter.close(ParquetRecordWriter.java:164)
at
org.apache.spark.sql.execution.datasources.parquet.ParquetOutputWriter.close(ParquetOutputWriter.scala:41)
at
org.apache.spark.sql.execution.datasources.FileFormatDataWriter.releaseCurrentWriter(FileFormatDataWriter.scala:74)
at
org.apache.spark.sql.execution.datasources.FileFormatDataWriter.releaseResources(FileFormatDataWriter.scala:129)
at
org.apache.spark.sql.execution.datasources.FileFormatDataWriter.commit(FileFormatDataWriter.scala:159)
at
org.apache.spark.sql.execution.datasources.SingleDirectoryDataWriter.commit(FileFormatDataWriter.scala:240)
at
org.apache.spark.sql.execution.datasources.FileFormatWriter$.$anonfun$executeTask$1(FileFormatWriter.scala:453)
at
org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1397)
at
org.apache.spark.sql.execution.datasources.FileFormatWriter$.executeTask(FileFormatWriter.scala:459)
... 17 more
Root cause: the internal Caffeine cache `internalFileSystemCache` calls
`FileSystem.close()` in
its eviction listener (size-based / time-based). If thread A is currently
using the cached
`FileSystem` instance while thread B triggers an eviction (cache full,
`expireAfterAccess` fired,
or `invalidate()` from a refresh path), the underlying FileSystem is closed
mid-call, breaking
thread A's in-flight I/O.
This is a classic "use-after-free" pattern on a pooled resource without
reference counting.
### How to reproduce
+ Which Gravitino version to use: `main` (reproduced on commit <1.0.0>),
also affects all released versions where `internalFileSystemCache` calls
`fs.close()`
in the eviction listener.
Reproduction steps (concurrent eviction race):
1. Build a `GravitinoVirtualFileSystem` instance and configure a small cache:
- fs.gravitino.fileset.cache.maxCapacity = 2
- fs.gravitino.fileset.cache.expireAfterAccess = 1s
(or use the default; both expiry and capacity-based eviction reproduce
the issue)
2. From multiple threads, repeatedly access more than 2 distinct filesets so
that the
cache is constantly evicting entries, e.g.:
ExecutorService pool = Executors.newFixedThreadPool(8);
for (int i = 0; i < 1000; i++) {
final int idx = i % 5; // 5 distinct filesets, cache capacity = 2 ->
frequent eviction
pool.submit(() -> {
Path p = new Path("gvfs://fileset/catalog/schema/fs_" + idx +
"/data.txt");
try (FSDataInputStream in = gvfs.open(p)) {
in.read(new byte[1024]);
}
});
}
3. Within seconds you will observe `IOException: Filesystem closed` from
threads whose
cached FileSystem was concurrently evicted (and `close()`d) by another
thread's lookup.
Expected behavior:
- An in-flight FileSystem operation must never see the underlying FileSystem
closed by the cache.
- Eviction should defer the actual `close()` until no caller is using the
instance
(reference-counted close).
Actual behavior:
- Eviction listener calls `fs.close()` immediately, racing with concurrent
users
-> `Filesystem closed` / `RejectedExecutionException`.
Fix proposal:
- Wrap each cached `FileSystem` in a reference-counted holder
(acquire on cache hit, release after use).
- Eviction listener marks the holder as "evicted" but only physically closes
when
refCount drops to 0.
- Provide a config switch `fs.gravitino.fileset.cache.closeOnEviction`
(default `false` = safe ref-counted close, `true` = legacy behavior) for
backward compatibility.
I have a working fix locally and will open a PR linking to this issue.
### Additional context
Affected component: clients/filesystem-hadoop3
(`internalFileSystemCache`).
Will be fixed by an upcoming PR introducing a reference-counted FileSystem
holder.
--
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]
