[FLINK-2491] Support Checkpoints After Tasks Finished

• Type: New Feature
• Status: Closed
• Priority: Major
• Resolution: Implemented
• Affects Version/s: 0.10.0
• Fix Version/s: 1.14.0
• Component/s: Runtime / Checkpointing
• Assignee: Yun Gao
• Reporter: Robert Metzger
• Created: 06/Aug/15 09:50
• Updated: 12/Apr/22 13:02
• Resolved: 21/Sep/21 09:18

Description

While implementing a test case for the Kafka Consumer, I came across the following bug:

Consider the following topology, with the operator parallelism in parentheses:

Source (2) --> Sink (1).

In this setup, the snapshotState() method is called on the source, but not on the Sink.
The sink receives the generated data.
only one of the two sources is generating data.

I've implemented a test case for this, you can find it here: https://github.com/rmetzger/flink/blob/para_checkpoint_bug/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ParallelismChangeCheckpoinedITCase.java

Issue Links

• causes: FLINK-23829 SavepointITCase JVM crash on azure
• causes: FLINK-24161 Can not stop the job with savepoint while a task is finishing
• is duplicated by: FLINK-13270 CheckpointCoordinator abort the checkpoint
• is duplicated by: FLINK-9640 Checkpointing is aways aborted if any task has been finished
• relates to: FLINK-21839 SinkFunction snapshotState don't snapshot all data when trigger a stop-with-drain savepoint
• relates to: FLINK-27202 NullPointerException on stop-with-savepoint with AsyncWaitOperator followed by FlinkKafkaProducer
• relates to: FLINK-22764 Disable Values Source node in streaming plan optimization
• relates to: FLINK-23883 (1.15) Follow up tasks and code improvements for FLIP-147
• mentioned in: Page Loading...

Sub-Tasks

• Refactor the CheckpointCoordinator to compute the tasks to trigger/wait/commit dynamically: Closed
• Modify the logic of computing tasks to trigger/wait/commit to consider finished tasks: Closed
• Stores finished status for fully finished operators: Closed
• Decline Checkpoint if some tasks finished before get triggered: Closed
• Allows tasks to report finish state: Closed
• Handle UnionListState with finished operators: Closed
• Handle BroadcastState with finished operators: Closed
• Refactor StreamTask hierarchy to support triggering checkpoint via RPC for non-source tasks: Closed
• Support checkpoint alignment with finished input channels: Closed
• StreamTask waits for all the asynchronous step of pending checkpoint to finish: Closed
• CheckpointCoordinator pass the flag about whether a operator is fully finished on recovery: Closed
• Skip the execution of the fully finished operators after recovery: Closed
• Re-compute tasks to trigger when tasks get triggered before finished: Closed
• Check for illegal modifications of JobGraph with finished operators: Closed
• Check for illegal modifications of JobGraph with partially finished operators: Closed
• Deprecate/Remove StreamOperator#dispose method: Closed
• Add finish method to the SinkFunction: Closed
• Call StreamOperator#finish on EndOfData: Closed
• Wait for a checkpoint completed after finishing a task: Closed
• Add a global flag for enabling/disabling final checkpoints: Closed
• Do not create transaction in TwoPhaseCommitSinkFunction after finish(): Closed
• Trigger global failover for synchronous savepoints on CheckpointCoordinator: Closed
• Pass watermarks in finishedOnRestore tasks: Closed
• When task is finishedOnRestore Operators shouldn't be used: Closed
• notifyCheckpointComplete without corresponding snapshotState: Closed
• FLIP-147 Checkpoint N has not been started at all: Closed
• Bypass operators when advanceToEndOfEventTime for both legacy and new source tasks: Closed
• FLIP-147: Unable to recover after source fully finished: Closed
• Take snapshots for operator coordinators which all corresponding tasks finished: Closed
• Fix the concurrent problem between triggering savepoint with drain and finish: Closed
• Double check if non-keyed FullyFinishedOperatorState can be mixed up with non finished OperatorState on recovery: Closed
• Do not resume channels if the barrier is received via RPC: Closed
• Respect the finished flag when extracting operator states due to skip in-flight data: Closed
• Upgrade the TwoPhaseCommitSink to support empty transaction after finished: Closed
• CheckpointBarrierHandler may skip the markAlignmentStart for alignment -with-timeout checkpoint: Closed
• Verify the final checkpoint and stop-with-savepoint --drain: Closed
• Add E2E/ITCase test for checkpoints after tasks finished: Closed
• Document FLIP-147 capabiliites and limitations: Closed
• Log improvement for aborting checkpoint due to tasks are finishing: Closed