private static CompletableFuture<Checkpoint> retryFailedRestore(boolean shouldRetryRestore, TaskName taskName,
      Map<TaskName, Checkpoint> taskCheckpoints, CheckpointManager checkpointManager,
      TaskRestoreManager taskRestoreManager, Config config, Map<TaskName, TaskInstanceMetrics> taskInstanceMetrics,
      ExecutorService executor, File loggedStoreBaseDirectory, JobContext jobContext, ContainerModel containerModel,
      String factoryName, Map<TaskName, Map<String, Set<String>>> taskBackendFactoryToStoreNames) {
    if (shouldRetryRestore) {
      // Try to restore with getDeleted flag - return a new task checkpoint
      return restoreDeletedSnapshot(taskName, taskCheckpoints, checkpointManager, taskRestoreManager, config,
          taskInstanceMetrics, executor, taskBackendFactoryToStoreNames.get(taskName).get(factoryName),
          loggedStoreBaseDirectory, jobContext, containerModel);
    }
    // if shouldRetryRestore is false, do not retry restore. This means restore completed successfully in the first try
    // Return the old task checkpoint as no new checkpoint was created
    return CompletableFuture.completedFuture(taskCheckpoints.get(taskName));
  }
Creates a future for either the new task checkpoint (if the restore was retried), or the old task checkpoint.
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerRestoreUtil.java