seungjoo-choi-bucketplace opened a new pull request, #29169: URL: https://github.com/apache/flink/pull/29169
## What is the purpose of the change Fixes [FLINK-40645](https://issues.apache.org/jira/browse/FLINK-40645): `FileCacheEntry#open` registers every `CachedDataInputStream` it creates in `FileCacheEntry#openedStreams`, but the only place that removes entries is `FileCacheEntry#doRemoveFile()`, i.e. when the file is evicted from or deleted in the cache. While a file stays cached, every open/close cycle of a stream on it leaves a closed stream object — together with the remote stream and its statistics objects it wraps — in the queue. Streams are opened and closed repeatedly when the RocksDB table cache reopens SST files (`state.backend.forst.files.open` small relative to the number of live files), when the per-file stream pool overflows, or when a read fails. In production (Flink 2.1.3, ForSt on S3A, `files.open=64`) this accumulated 83,403 `S3AInputStream` instances (~650 MB) in 24 minutes and the TaskManagers died of heap exhaustion. ## Brief change log - `CachedDataInputStream#close`: remove the stream from `cacheEntry.openedStreams` as the first step, so that a failing close cannot skip the removal. `LinkedBlockingQueue#remove` is safe against the concurrent iteration in `doRemoveFile()`. - New `FileCacheEntryTest`. ## Verifying this change This change added tests and can be verified as follows: - `FileCacheEntryTest#testClosedStreamIsUnregisteredFromEntry`: two streams opened through `FileCacheEntry#open` are registered; closing one removes exactly that one (idempotently), closing the other empties the queue. Fails on master (`closed stream must not stay registered`). - Production observation: with the fix (and `files.open=-1`) the number of `S3AInputStream` instances stays proportional to the number of open files (468 instances for 67 files after 18 hours). ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): no - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- FLINK-40644 fixes the other half of the same stream life cycle (the remote stream not being closed) and touches the same method; the two changes are independent and merge cleanly in either order. The analysis and the patch were prepared with AI assistance (Claude Code) and validated in production as described above. 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
