zhijiangW commented on a change in pull request #11509:
URL: https://github.com/apache/flink/pull/11509#discussion_r496557586
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/AsyncCheckpointRunnable.java
##########
@@ -197,7 +199,8 @@ private void handleExecutionException(Exception e) {
// We only report the exception for the
original cause of fail and cleanup.
// Otherwise this followup exception could race
the original exception in failing the task.
try {
-
taskEnvironment.declineCheckpoint(checkpointMetaData.getCheckpointId(),
checkpointException);
+
taskEnvironment.declineCheckpoint(checkpointMetaData.getCheckpointId(),
+ new
CheckpointException(CheckpointFailureReason.EXCEPTION, checkpointException));
Review comment:
> CheckpointFailureReason.JOB_FAILURE is the default value of method
getCheckpointException, you can check the implementation for more details.
Yeah, you are right.
> You're right about the semantic. In our internal version we add a new
CheckpointFailureReason value:
I like the `CHECKPOINT_ASYNC_EXCEPTION` way. Go ahead!
BTW, how about supplementing somehow unit test for the changes if possible?
Maybe construct `AsyncCheckpointRunnable` like did in
`LocalStateForwardingTest` and then throw an exception during `run`. Then we
can override the implementation of `StreamMockEnvironment#declineCheckpoint` to
verify the exception argument `instanceof CheckpointException` and `preFlight =
false`. Or there are some other easier way I did not think through.
----------------------------------------------------------------
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]
