@Override
  public void init(Checkpoint checkpoint) {
    cleanBaseDirsAndReadOffsetFiles();
    setupBaseDirs();
    validateChangelogStreams();
    getOldestChangeLogOffsets();
    registerStartingOffsets();
  }
Cleans up and sets up store directories, validates changeLog SSPs for all stores of this task,
and registers SSPs with the respective consumers.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java