rkhachatryan commented on a change in pull request #18391: URL: https://github.com/apache/flink/pull/18391#discussion_r787623075
########## File path: flink-runtime/src/main/java/org/apache/flink/runtime/state/changelog/ChangelogStateBackendHandle.java ########## @@ -66,16 +71,36 @@ public ChangelogStateBackendHandleImpl( @Override public void registerSharedStates(SharedStateRegistry stateRegistry, long checkpointID) { + for (KeyedStateHandle keyedStateHandle : materialized) { + registerState(stateRegistry, checkpointID, MATERIALIZED_FLAG, keyedStateHandle); + } + for (ChangelogStateHandle stateHandle : nonMaterialized) { + registerState(stateRegistry, checkpointID, NON_MATERIALIZED_FLAG, stateHandle); + } stateRegistry.registerAll(materialized, checkpointID); stateRegistry.registerAll(nonMaterialized, checkpointID); } + private void registerState( + SharedStateRegistry stateRegistry, + long checkpointID, + String prefix, + KeyedStateHandle keyedStateHandle) { + stateRegistry.registerReference( + new SharedStateRegistryKey( + prefix, + // here use hash code as registry key identifier. + new StateHandleID(String.valueOf(keyedStateHandle.hashCode()))), + new StreamStateHandleWrapper(keyedStateHandle), + checkpointID); Review comment: I think resetting Writer to a given SQN is possible with the current DFS changelog implementation, but this may be problematic in other implementations. Using SQN also implies that the `changelogWriter.persist(sqn)` must return a new handle for each new SQN, even if nothting was changed (SQN can generally increment without state changes). Unless the returned handle doesn't contain any "private" state (which is currently the case), but then we don't need to worry about non-materialized state at all. -- 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