public Map<TaskName, Checkpoint> start() throws SamzaException, InterruptedException {
    // Restores and recreates stores.
    Map<TaskName, Checkpoint> taskCheckpoints = restoreStores();

    // Shutdown restore executor since it will no longer be used
    try {
      restoreExecutor.shutdown();
      if (restoreExecutor.awaitTermination(RESTORE_THREAD_POOL_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.MILLISECONDS)) {
        restoreExecutor.shutdownNow();
      }
    } catch (Exception e) {
      LOG.error("Error shutting down restore executor", e);
    }

    // create and restore side input stores
    this.sideInputsManager = new SideInputsManager(
        sideInputSystemStreams, systemFactories,
        changelogSystemStreams, activeTaskChangelogSystemStreams,
        storageEngineFactories, containerModel, jobContext, containerContext,
        samzaContainerMetrics, taskInstanceMetrics, taskInstanceCollectors,
        streamMetadataCache, systemAdmins, serdeManager, serdes, storageManagerUtil,
        loggedStoreBaseDirectory, nonLoggedStoreBaseDirectory, config, clock);

    // blocks until initial side inputs restore is complete
    sideInputsManager.start();

    // Refactor Note (prateekm): In the previous version, side input stores were created in constructor,
    // and were added to taskStores (which are created in restoreStores()) at the end of restoreStores().
    // In this version, side input stores are created much later during start() (this method),
    // in the SideInputsManager constructor call above, and are added to taskStores here after side input restore
    // (in sideInputsManager.start()) is complete. Completing the side input restore first isn't strictly necessary.
    sideInputsManager.getSideInputStores().forEach((taskName, stores) -> {
      if (!this.taskStores.containsKey(taskName)) {
        taskStores.put(taskName, new HashMap<>());
      }
      taskStores.get(taskName).putAll(stores);
    });

    isStarted = true;
    return taskCheckpoints;
  }
Starts all the task stores.
Returns the latest checkpoint for each task. This checkpoint may be different from the lastCheckpoint returned by
checkpoint manager in case of a BlobStoreRestoreManager.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManager.java