Currently, restoring the latest checkpoint always recovers the CompletedCheckpointStore in CheckpointCoordinator::restoreLatestCheckpointedState(...) by calling completedCheckpointStore.recover(sharedStateRegistry). I think this is not required on every restore, but can happen only once when the JM becomes leader, e.g. when the CheckpointCoordinator is created. This can avoid unnecessary work and simplify the logic, e.g. for populating the SharedStateRegistry, and we could probably also move the ownership for SharedStateRegistry into some AbstractCompletedCheckpontStore.