[FLINK-39140][checkpoint] Allow multiple rescales in Unaligned Checkpoint ITCases to perform checkpointing during recovery

[1996fanrui]

What is the purpose of the change

Current Unaligned Checkpoint ITCases only restart once from a normal checkpoint. They do not cover restoring from a checkpoint produced by recovery phase — which is the key scenario for checkpointing during recovery.

Proposed mechanism: After restoring from a checkpoint, wait for the first new checkpoint to be produced, then immediately trigger a restart from it. Repeat for a configurable number of rounds (≥ 2). Whether to rescale depends on the specific test case.

This mechanism works on the current master (validating normal checkpoint recovery). Once checkpointing during recovery is enabled, the same tests automatically cover recovery-phase checkpoint scenarios.

Brief change log

• [FLINK-39140][checkpoint] Disable CUSTOM_PARTITIONER in unaligned checkpoint it case since it does not work well
  • See FLINK-39162
• [FLINK-39140][checkpoint] Allow multiple rescales in Unaligned Checkpoint ITCases to perform checkpointing during recovery
• [FLINK-39140][checkpoint] Fix MAX_RETAINED_CHECKPOINTS not effective in UnalignedCheckpointRescaleWithMixedExchangesITCase
  • Move MAX_RETAINED_CHECKPOINTS from per-job config to MiniCluster cluster config. StandaloneCompletedCheckpointStore reads this value from the cluster-level configuration, so setting it in StreamExecutionEnvironment had no effect (defaulting to 1).
  • This caused checkpoint is cleaned up sometimes due to new checkpoint is generated
• [FLINK-39140][checkpoint] Change record type from Long to String in UnalignedCheckpointRescaleWithMixedExchangesITCase
  • Long records (8 bytes) allow hundreds of records per buffer, causing excessive backpressure during aligned checkpoint phases (forward/rescale exchanges). Using 100-char random String records reduces the record count per buffer, shortening the time needed to drain backpressured buffers.
• [hotfix] Including task name and subtask index into channel-state-unspilling thread name

Verifying this change

This is an enhancement for existing ITCase, so it has been covered by testing.

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

Documentation

• Does this pull request introduce a new feature? no

[flinkbot]

CI report:

• f6da7ce Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[rkhachatryan]

Thanks for the PR @1996fanrui

I've left a couple of comments, PTAL.

Apart from that, could you add component name to the last commit message (I guess [runtime])?
And for the earlier commits, [test] seem to be a more relevant component.

[1996fanrui]

Hey @rkhachatryan , thanks for the review!

Apart from that, could you add component name to the last commit message (I guess [runtime])? And for the earlier commits, [test] seem to be a more relevant component.

Sounds make sense, updated.

[rkhachatryan]

Thanks for updating the PR, LGTM.

W.r.t. CustomPartitioner (FLINK-39162), I've drafted #27731 - my guess is that we don't forward disableUnalignedCheckpoints call to the nested partitioner (I didn't look deeply into the code).