public void run() {
    setup();

    LOG.info("start recovering...");

    systemAdmins.start();
    this.containerStorageManagers.forEach((containerName, containerStorageManager) -> {
      try {
        containerStorageManager.start();
      } catch (InterruptedException e) {
        // we can ignore the exception since its only used in the context of a command line tool and bubbling the
        // exception upstream isn't needed.
        LOG.warn("Received an interrupt during store restoration for container {}."
            + " Proceeding with the next container", containerName);
      }
    });
    this.containerStorageManagers.forEach((containerName, containerStorageManager) -> containerStorageManager.shutdown());
    systemAdmins.stop();

    LOG.info("successfully recovered in " + storeBaseDir.toString());
  }
run the setup phase and restore all the task storage
samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java