[FLINK-38408][checkpoint] Complete the checkpoint CompletableFuture after updating statistics to ensures semantic correctness and prevent test failure

[1996fanrui]

What is the purpose of the change

MapStateNullValueCheckpointingITCase failed with No checkpoint was created yet

Root Cause Analysis

Problem Location

Log analysis revealed that the checkpoint had actually completed successfully:

07:19:37,522 [jobmanager-io-thread-1] INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator [] - Completed checkpoint 1 for job b809cf46d67c23697786fd514565c737 (4464 bytes, checkpointDuration=45 ms, finalizationTime=4 ms)

However, the test code could not find the completed checkpoint when calling CommonTestUtils.getLatestCompletedCheckpointPath().

Root Cause

The problem occurs in the execution order of the CheckpointCoordinator.completePendingCheckpoint() method:

flink/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java

Line 1389 in 39a4628

reportCompletedCheckpoint(completedCheckpoint);

pendingCheckpoint.getCompletionFuture().complete(completedCheckpoint);
reportCompletedCheckpoint(completedCheckpoint);

Checkpoint Coordinator mechanism:

1. ​A​: pendingCheckpoint.getCompletionFuture().complete(completedCheckpoint) completes the completion future first{}
2. ​B​: reportCompletedCheckpoint(completedCheckpoint) updates checkpoint statistics.

Test code timeline:

1. C: Detect future completion
2. D: Call getLatestCompletedCheckpointPath() immediately

Usually, the execution sequence is A -> B -> C -> D, it works well.

The bug happens if execution sequence is A > C -> D -> B.

Reproduction Method

In the completePendingCheckpoint() method, inserting Thread.sleep(100) between complete() and reportCompletedCheckpoint() can reproduce this issue 100%.

Brief change log: Adjust the execution order in CheckpointCoordinator

[FLINK-38408][checkpoint] Complete the checkpoint CompletableFuture after updating statistics to ensures semantic correctness and prevent test failure

Changes:

// Update statistics first 
reportCompletedCheckpoint(completedCheckpoint);
// Complete the future later
pendingCheckpoint.getCompletionFuture().complete(completedCheckpoint);

Benefits:

• Fundamentally eliminates race conditions
• Ensures semantic correctness: Waiting parties are notified only when the checkpoint is fully processed

Verifying this change

• Added testCompletionFutureCompletesAfterReporting

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:

• b0e8240 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[Izeren]

Thank you for the change @1996fanrui. Overall, LGTM, my main concern is about potential test flakiness, PTAL

[Izeren]

I have concerns that change like this can have potential impacts like:

1. Deadlock / race condition if reportCompletedCheckpoint would trigger any handler that also waits on the checkpoint future before its completion (in general, unlikely situation, and should be caught by existing test)
2. Checkpoint completion will be slightly delayed, but reporting is a quick operation, so doesn't seem to be critical
3. If reporting throws exception it will result in checkpoint being completed exceptionally. Could we confirm that this behaviour matches the previous one?

[Izeren]

That is likely to end up being flaky test. Test in CI could freeze for 15min and more, so 20 seconds timeout may not be sufficient in general.
I suggest to use indefinite timeout of at least a few hours

[Izeren]

Similarly to above, I am not sure you can confirm whether expected change did not occur because of being blocked vs corresponding thread being inactive. Will be better to wait indefinitely here

[Izeren]

ditto

[Izeren]

ditto