[FLINK-40644][state/forst] Close the remote stream when a CachedDataInputStream is closed

[seungjoo-choi-bucketplace]

What is the purpose of the change

Fixes 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 (#29169) 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

[flinkbot]

CI report:

• 5f86fe1 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build