[FLINK-10753] Propagate and log snapshotting exceptions

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.6.2, 1.7.0
• Fix Version/s: 1.5.6, 1.6.3, 1.7.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Assignee: Stefan Richter
• Reporter: Alexander Fedulov
• Created: 01/Nov/18 15:35
• Updated: 20/Dec/18 09:27
• Resolved: 12/Nov/18 17:39

Description

Upon failure, AbstractStreamOperator.snapshotState rethrows a new exception with the message "Could not complete snapshot {} for operator {}." and the original exception as the cause. 

While handling the error, CheckpointCoordinator.discardCheckpoint method logs only this  propagated message and not the original cause of the exception.

In addition, pendingCheckpoint.abortDeclined(), called from the discardCheckpoint, reports the failed checkpoint with a misleading message "Checkpoint was declined (tasks not ready)". This message is what will be displayed in the UI (see attached).

 Proposition:

1. Log exception at the Task Manager (.snapshotState)
2. Log cause, instead of cause.getMessage() at the JobsManager (.dicardCheckpoint)
3. Pass root cause to abortDeclined and propagate a more appropriate message to the UI.

Issue Links

• is duplicated by: FLINK-11190 Log stack trace on declined/discarded checkpoint
• links to: GitHub Pull Request #7064