[FLINK-13593] Prevent failing the wrong execution attempt in CheckpointFailureManager

• Type: Bug
• Status: Resolved
• Priority: Blocker
• Resolution: Fixed
• Affects Version/s: 1.9.0
• Fix Version/s: 1.9.0
• Component/s: Runtime / Checkpointing
• Labels: pull-request-available
• Assignee: Yu Li
• Reporter: Yu Li
• Created: 06/Aug/19 05:55
• Updated: 12/Aug/19 09:59
• Resolved: 09/Aug/19 12:50

Description

Due to the asynchronously handling of checkpoint decline message in LegacyScheduler#declineCheckpoint, it's possible that the message is handled before job status transition thus receiveDeclineMessage grabbed the lock in CheckpointCoordinator before pendingCheckpoints got cleared by stopCheckpointScheduler (as triggered by the job status listener CheckpointCoordinatorDeActivator). And if the job/tasks restarts quickly enough, the FailJobCallback in CheckpointFailureManager might unexpectedly fail the job again, as observed in FLINK-13527.

To resolve the issue, we need to add a safe guard when failing the job, passing through the ExecutionAttemptID and checking against the current executions to make sure the to-be-failed one is still running, so we won't fail the newly restarted one by accident.

Issue Links

• causes: FLINK-13527 Instable KafkaProducerExactlyOnceITCase due to CheckpointFailureManager
• is caused by: FLINK-13695 Integrate checkpoint notifications into StreamTask's lifecycle
• is duplicated by: FLINK-13527 Instable KafkaProducerExactlyOnceITCase due to CheckpointFailureManager
• links to: GitHub Pull Request #9364