public static CompletableFuture<Map<TaskName, Checkpoint>> initAndRestoreTaskInstances(
      Map<TaskName, Map<String, TaskRestoreManager>> taskRestoreManagers, SamzaContainerMetrics samzaContainerMetrics,
      CheckpointManager checkpointManager, JobContext jobContext, ContainerModel containerModel,
      Map<TaskName, Checkpoint> taskCheckpoints, Map<TaskName, Map<String, Set<String>>> taskBackendFactoryToStoreNames,
      Config config, ExecutorService executor, Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
      File loggerStoreDir, Map<String, SystemConsumer> storeConsumers) {

    Set<String> forceRestoreTasks = new HashSet<>();
    // Initialize each TaskStorageManager.
    taskRestoreManagers.forEach((taskName, restoreManagers) ->
        restoreManagers.forEach((factoryName, taskRestoreManager) -> {
          try {
            taskRestoreManager.init(taskCheckpoints.get(taskName));
          } catch (SamzaException ex) {
            if (isUnwrappedExceptionDeletedException(ex) && taskRestoreManager instanceof BlobStoreRestoreManager) {
              // Get deleted SnapshotIndex blob with GetDeleted and mark the task to be restored with GetDeleted as well.
              // this ensures that the restore downloads the snapshot, recreates a new snapshot, uploads it to blob store
              // and creates a new checkpoint.
              ((BlobStoreRestoreManager) taskRestoreManager).init(taskCheckpoints.get(taskName), true);
              forceRestoreTasks.add(taskName.getTaskName());
            } else {
              // log and rethrow exception to communicate restore failure
              String msg = String.format("init failed for BlobStoreRestoreManager, task: %s with GetDeleted set to true", taskName);
              LOG.error(msg, ex);
              throw new SamzaException(msg, ex);
            }
          }
        })
    );

    // Start each store consumer once.
    // Note: These consumers are per system and only changelog system store consumers will be started.
    // Some TaskRestoreManagers may not require the consumer to be started, but due to the agnostic nature of
    // ContainerStorageManager we always start the changelog consumer here in case it is required
    storeConsumers.values().stream().distinct().forEach(SystemConsumer::start);

    return restoreAllTaskInstances(taskRestoreManagers, taskCheckpoints, taskBackendFactoryToStoreNames, jobContext,
        containerModel, samzaContainerMetrics, checkpointManager, config, taskInstanceMetrics, executor, loggerStoreDir,
        forceRestoreTasks);
  }
Inits and Restores all the task stores.
Note: In case of {@link BlobStoreRestoreManager}, this method retries init and restore with getDeleted flag if it
receives a {@link DeletedException}. This will create a new checkpoint for the corresponding task.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerRestoreUtil.java