seungjoo-choi-bucketplace opened a new pull request, #29168: URL: https://github.com/apache/flink/pull/29168
## What is the purpose of the change Fixes [FLINK-40644](https://issues.apache.org/jira/browse/FLINK-40644): with the ForSt file cache, `CachedDataInputStream.close()` only closes the cached (local) stream and never closes the original remote stream it wraps. `FileCacheEntry#open` hands the remote stream over to the wrapper and keeps no other reference, so nothing ever closes it. Every remote stream that has read from S3 holds a leased connection of the S3A HTTP connection pool, and the pool is not GC-aware, so each close leaks one connection for the lifetime of the TaskManager. Once the pool is exhausted all reads block in `AbstractConnPool.getPoolEntryBlocking` and the job fails with `Timeout waiting for connection from pool` after `s3.connection.timeout`, restarts, and repeats. Streams are closed while their file is still cached whenever the RocksDB table cache evicts an SST (`state.backend.forst.files.open` smaller than the number of live files), when compaction input files are deleted, and when the per-file stream pool of `ByteBufferReadableFSDataInputStream` overflows. `readFully(long, ByteBuffer)` has a related gap: a pooled stream that throws while reading is neither returned to the pool nor closed. ## Brief change log - `CachedDataInputStream#close`: close the original stream as well; if both the cached and the original stream fail to close, propagate the first exception with the second attached as suppressed. - `ByteBufferReadableFSDataInputStream#readFully(long, ByteBuffer)`: close the pooled stream when the positioned read fails. - New `CachedDataInputStreamTest` and `ByteBufferReadableFSDataInputStreamTest`. ## Verifying this change This change added tests and can be verified as follows: - `CachedDataInputStreamTest`: a stream opened through `FileCacheEntry#open` closes its original stream on `close()` (idempotently); when both a cached and an original stream are present both are closed; when the cached stream fails to close the original stream is still closed and the second exception is suppressed. The first test fails on master (`original stream must be closed`). - `ByteBufferReadableFSDataInputStreamTest`: a pooled stream whose positioned read throws is closed by `readFully`. Fails on master. - Production observation (Flink 2.1.3, ForSt on S3A, cache smaller than a ~40 GB TTL state): before the fix the number of ESTABLISHED connections to S3 per TaskManager grew monotonically with the number of closed streams and the job hung after ~6 h with `s3.connection.maximum=96` (~2 days with 1024); with the fix deployed as a class override the count stays in the single digits across hundreds of thousands of cache misses. ## 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 (the change is in the ForSt file cache; the effect is on how many remote streams it keeps open) ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable --- FLINK-40645 fixes the other half of the same stream life cycle (closed streams staying registered in `FileCacheEntry#openedStreams`) and touches the same method; the two changes are independent. 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]
