[FLINK-16753] Exception from AsyncCheckpointRunnable should be wrapped in CheckpointException

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.10.0, 1.11.0, 1.11.1
• Fix Version/s: 1.10.3, 1.11.3, 1.12.0
• Component/s: Runtime / Checkpointing
• Labels: pull-request-available usability
• Assignee: Jiayi Liao
• Reporter: Jiayi Liao
• Created: 24/Mar/20 16:28
• Updated: 16/Nov/20 09:31
• Resolved: 13/Oct/20 06:11

Description

If an exception is thrown from task's async checkpoint process, the checkpoint will be declined as expected, but the reason for declining checkpoint will be regarded as CheckpointFailureReason.JOB_FAILURE, which gives a wrong message to users.

I think we can simply replace

owner.getEnvironment().declineCheckpoint(checkpointMetaData.getCheckpointId(), checkpointException);

with

owner.getEnvironment().declineCheckpoint(checkpointMetaData.getCheckpointId(), new CheckpointException(CheckpointFailureReason.EXCEPTION, checkpointException));

in AsyncCheckpointRunnable.handleExecutionException.

cc Till Rohrmann

Issue Links

• is duplicated by: FLINK-20099 HeapStateBackend checkpoint error hidden under cryptic message
• relates to: FLINK-18962 Improve error message if checkpoint directory is not writable
• links to: GitHub Pull Request #11509
• links to: GitHub Pull Request #14071
• links to: GitHub Pull Request #14072