[GitHub] [flink] gaoyunhaii commented on a change in pull request #16913: [FLINK-23809][checkpoint] Respect the finished flag when extracting operator states due to skip in-flight data

[GitBox]

gaoyunhaii commented on a change in pull request #16913:
URL: https://github.com/apache/flink/pull/16913#discussion_r694071016



##########
File path: 
flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
##########
@@ -1601,10 +1601,15 @@ private OptionalLong 
restoreLatestCheckpointedStateInternal(
         // Create the new operator states without in-flight data.
         for (OperatorState originalOperatorState : 
originalOperatorStates.values()) {
             OperatorState newState =
-                    new OperatorState(
-                            originalOperatorState.getOperatorID(),
-                            originalOperatorState.getParallelism(),
-                            originalOperatorState.getMaxParallelism());
+                    originalOperatorState.isFullyFinished()
+                            ? new FullyFinishedOperatorState(
+                                    originalOperatorState.getOperatorID(),
+                                    originalOperatorState.getParallelism(),
+                                    originalOperatorState.getMaxParallelism())
+                            : new OperatorState(
+                                    originalOperatorState.getOperatorID(),
+                                    originalOperatorState.getParallelism(),
+                                    originalOperatorState.getMaxParallelism());

Review comment:
       Very thanks for the suggestions! It makes a lot of senses to me~ I 
modified the code accordingly~




-- 
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.

To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org