[FLINK-33090] CheckpointsCleaner clean individual checkpoint states in parallel

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.17.1
• Fix Version/s: 1.19.0
• Component/s: Runtime / Checkpointing
• Labels: pull-request-available

Release Note

Now when disposing of no longer needed checkpoints, every state handle/state file will be disposed in parallel by the ioExecutor, vastly improving disposing speed of a single checkpoint (for large checkpoints the disposal time can be improved from 10 minutes to < 1 minute) . The old behaviour can be restored by setting `state.checkpoint.cleaner.parallel-mode` to `false`.

• Assignee: Yi Zhang
• Reporter: Yi Zhang
• Created: 14/Sep/23 17:39
• Updated: 25/Oct/23 18:25
• Resolved: 25/Oct/23 14:00

Description

Currently CheckpointsCleaner clean multiple checkpoints in parallel with JobManager's ioExecutor, however each checkpoint states is cleaned sequentially. With thousands of StateObjects to clean this can take long time on some checkpoint storage, if longer than the checkpoint interval this prevents new checkpointing.

The proposal is to use the same ioExecutor to clean up each checkpoints states in parallel as well. From my local testing, with default settings for ioExecutor thread pool for xK state files this can reduce clean up time from 10 minutes to <1 minute. 

Issue Links

• relates to: FLINK-26590 Triggered checkpoints can be delayed by discarding shared state
• links to: GitHub Pull Request #23425