[FLINK-37379][state/forst] Double list LRU in ForSt state and cache reload

[Zakelly]

What is the purpose of the change

Currently, the file-based LRU cache lacks of loading files from the disaggregated DFS. Meaning that after the job recovery, there will be no cache available, hence the performance is low over a period of time. It is critical for production availability of disaggregated state.

This PR introduces a new implementation that allows ForSt to manage file-based cache in a double list (hot/cold) and support cache reload after eviction.

Brief change log

• Introduce DoubleListLru and rewrite the FileBasedCache based on that.

Verifying this change

This change is already covered by existing IT tests. The new introduced classes are covered by new introduced UTs.

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): yes
• 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? yes
• If yes, how is the feature documented? docs

[flinkbot]

CI report:

• 6fe6ad5 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[fredia]

Thanks for the PR, overall design LGTM, left two comments regarding the implementation.

[fredia]

Is the FileCacheEntry#cachePath deleted from cacheFs?
Do we need to wait for delete before releasing cacheLimitPolicy?

[Zakelly]

The cache will be deleted in value.invalidate(), if the reference count reached 0.

We'd better not wait for that, since we are on the task thread and it's better not block here.

[Zakelly]

Also, additional two commits to enhance the user experience are attached.