[FLINK-40685][state/forst] Reclaim cached files after checkpoint ownership handoff

[minxhe]

What is the purpose of the change

Release ForSt's local cached copy when a mapping reaches zero references, even if the source file is owned by a checkpoint. The checkpoint-owned source must remain untouched.

Jira: https://issues.apache.org/jira/browse/FLINK-40685

Brief change log

• Restrict the ownership guard to source deletion, allowing eligible cache cleanup for checkpoint-owned files while preserving owned-source cleanup order.
• Ignore touches of closed cache entries under the same lock used for removal, preventing stale LRU-node access.
• Complete the cached-to-original position handoff after failed acquisition of a closed entry, even while asynchronous removal is queued.
• Cover real FileStateHandle handoff, owned files, and correct-position open-reader fallback before and after physical cache removal on both Flink-marked and unmarked threads.

Verifying this change

Six new MappingEntryTest cases cover real file and cache lifecycle behavior. Three fail on unchanged Apache master c7f0069db4d95a4fdf5af0eb038cafffe8c0b2bb because cache invalidation is skipped. The Flink-marked reader also detects a stale-node failure with the ownership-only fix, establishing the need for the closed-entry guard.

The same two reader cases now hold the cache entry's removal monitor: read byte 1 from cache, release the mapping, read byte 2 while removal is blocked, then allow removal and read bytes 3 and 4. Without the position-handoff amendment, both return byte 1 instead of 2. No executor injection, worker-count assumptions or sleeps are needed to enforce this interleaving.

With the complete amended fix, all 338 focused cases pass, without failures, errors or skips:

./mvnw -B -ntp -pl flink-state-backends/flink-statebackend-forst \
  -Dflink.forkCountUnitTest=1 \
  -Dtest=MappingEntryTest,FileMappingManagerTest,ForStFlinkFileSystemTest,DoubleListLruTest test

Using Maven 3.9.16 and JDK 17, module clean verify passed with the strengthened queued-removal cases and separate test JVMs: per-class XML reports record 984 passing invocations, including two integration invocations, and 54 existing skipped invocations, with no failures or errors. Spotless, Checkstyle and the module's API compatibility check passed.

./mvnw -B -ntp -pl flink-state-backends/flink-statebackend-forst \
  -Dflink.forkCountUnitTest=1 -Dflink.forkCountITCase=1 \
  -Dflink.reuseForks=false clean verify

The initial reused-fork module run exposed existing thread-local pollution: the unchanged ForStAsyncAndSyncCompatibilityTest#testForStTransFromAsyncToSync followed by ForStFlinkFileSystemTest#testSstFileInCache also fails on unchanged upstream production. The same existing pair passes with separate test JVMs. No existing tests were changed or excluded.

Full-reactor clean verify has not been run.

Does this pull request potentially affect one of the following parts:

• Dependencies: no
• Public API: no
• Serializers: no
• Runtime per-record code paths: yes, a closed-entry check in the existing synchronized cache-touch callback and position handoff after unsuccessful cached-stream acquisition; successful cache reads and eviction/promotion policies are unchanged
• Deployment or recovery: yes, checkpoint file lifecycle
• S3 file system connector: no connector code change; source preservation must hold for remote files

Documentation

• New feature: no
• Documentation: ownership rationale in code; no configuration changes

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: GitHub Copilot CLI 1.0.84-5

[flinkbot]

CI report:

• a21f185 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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