[FLINK-2491] Checkpointing only works if all operators/tasks are still running

• Type: Bug
• Status: In Progress
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 0.10.0
• Component/s: Runtime / Checkpointing
• Assignee: Unassigned
• Reporter: Robert Metzger
• Created: 06/Aug/15 09:50
• Updated: 1 week ago 13:51

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

• 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
• 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: Open
• Decline Checkpoint if some tasks finished before get triggered: Open
• Allows tasks to report operator subtask finish state: Reopened
• Identify JobVertex containing legacy source operators and abort checkpoint with legacy source operators partially finished: Open
• Refactor StreamTask hierarchy to support triggering checkpoint via RPC for non-source tasks: Open
• CheckpointBarrierHandler Insert barriers for channels received EndOfPartition: Open
• StreamTask waits for all the asynchronous step of pending checkpoint to finish: Open
• CheckpointCoordinator pass the flag about whether a operator is fully finished on recovery: Open
• Skip the execution of the fully finished operators after recovery: Open
• Enable Checkpoints after tasks finished: Open
• Re-compute tasks to trigger when tasks get triggered before finished: Closed