[GitHub] [flink] akalash commented on a diff in pull request #19331: [FLINK-26985][runtime] Don't discard shared state of restored checkpoints

[GitBox <gi...@apache.org>]

akalash commented on code in PR #19331:
URL: https://github.com/apache/flink/pull/19331#discussion_r843801124


##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistry.java:
##########
@@ -32,11 +33,11 @@
 
     /** A singleton object for the default implementation of a {@link SharedStateRegistryFactory} */
     SharedStateRegistryFactory DEFAULT_FACTORY =
-            (deleteExecutor, checkpoints) -> {
+            (deleteExecutor, checkpoints, restoreMode) -> {
                 SharedStateRegistry sharedStateRegistry =
                         new SharedStateRegistryImpl(deleteExecutor);
                 for (CompletedCheckpoint checkpoint : checkpoints) {
-                    checkpoint.registerSharedStatesAfterRestored(sharedStateRegistry);
+                    checkpoint.registerSharedStatesAfterRestored(sharedStateRegistry, restoreMode);

Review Comment:
   Maybe does it make sense to create the method `registerSharedStates` without `restoreMode` and use it here? In my opinion, it makes sense since all usage of `DEFAULT_FACTORY` uses always DEFAULT restore mode. And secondly, in general, I think that this place is not about `AfterRestore` it is used at any time of lifecycle(but I can be mistaken here). The single place which, in my opinion, `AfterRestore` is `CheckpointCoordinator#restoreSavepoint`.
   This idea allows reverting half of your changes I think.
   WDYT?



##########
flink-runtime/src/main/java/org/apache/flink/runtime/state/SharedStateRegistryImpl.java:
##########
@@ -174,6 +181,20 @@ public void registerAll(
         }
     }
 
+    @Override
+    public void registerAllAfterRestored(CompletedCheckpoint checkpoint, RestoreMode mode) {

Review Comment:
   I don't really sure that `SharedStateRegistryImpl` should know something about `RestoreMode` and especially its connection to `highestRetainCheckpointID`. Maybe is it better to use `highestRetainCheckpointID` as the second parameter here instead of `RestoreMode`. And `CheckpointCoordinator#restoreSavepoint` can calculate this `highestRetainCheckpointID` based on the `RestoreMode` paramater which totally makes sense since `CheckpointCoordinator` know everything about restoring.
   WDYT?



-- 
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-unsubscribe@flink.apache.org

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