StefanRRichter commented on a change in pull request #8322: [FLINK-12364]
Introduce a CheckpointFailureManager to centralized manage checkpoint failure
URL: https://github.com/apache/flink/pull/8322#discussion_r292955466
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/messages/checkpoint/DeclineCheckpoint.java
##########
@@ -48,15 +43,24 @@ public DeclineCheckpoint(JobID job, ExecutionAttemptID
taskExecutionId, long che
public DeclineCheckpoint(JobID job, ExecutionAttemptID taskExecutionId,
long checkpointId, Throwable reason) {
super(job, taskExecutionId, checkpointId);
- if (reason == null ||
- reason.getClass() ==
AlignmentLimitExceededException.class ||
- reason.getClass() ==
CheckpointDeclineOnCancellationBarrierException.class ||
- reason.getClass() ==
CheckpointDeclineSubsumedException.class ||
- reason.getClass() ==
CheckpointDeclineTaskNotCheckpointingException.class ||
- reason.getClass() ==
CheckpointDeclineTaskNotReadyException.class ||
- reason.getClass() == InputEndOfStreamException.class) {
- // null or known common exceptions that cannot
reference any dynamically loaded code
+ if (reason == null) {
Review comment:
Why not instead just
```
if (reason == null || reason instanceof CheckpointException) {
this.reason = reason;
} else {
// some other exception. replace with a serialized
throwable, to be on the safe side
this.reason = new SerializedThrowable(reason);
}
```
All `CheckpointException` objects should be from the Flink code and not user
defined and therefore not require any `SerializedThrowable`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
