@Override
  public void init(Checkpoint checkpoint) {
    Map<String, KafkaStateCheckpointMarker> storeStateCheckpointMarkers = getCheckpointedChangelogOffsets(checkpoint);
    currentChangelogOffsets = getCurrentChangelogOffsets(taskModel, storeChangelogs, sspMetadataCache);

    this.storeActions = getStoreActions(taskModel, storeEngines, storeChangelogs,
        storeStateCheckpointMarkers, getCheckpointId(checkpoint), currentChangelogOffsets, systemAdmins, storageManagerUtil,
        loggedStoreBaseDirectory, nonLoggedStoreBaseDirectory, config, clock);

    setupStoreDirs(taskModel, storeEngines, storeActions, storageManagerUtil, fileUtil,
        loggedStoreBaseDirectory, nonLoggedStoreBaseDirectory);
    registerStartingOffsets(taskModel, storeActions, storeChangelogs, systemAdmins, storeConsumers,
        currentChangelogOffsets);
  }
This class implements the state restore based on state snapshots of checkpoints and changelog.
samza-core/src/main/java/org/apache/samza/storage/TransactionalStateTaskRestoreManager.java