private static void createCheckpointDirFromStoreDirCopy(TaskName taskName, TaskModel taskModel,
      File loggedStoreBaseDir, Set<String> storeName, CheckpointId checkpointId) {
    StorageManagerUtil storageManagerUtil = new StorageManagerUtil();
    for (String store : storeName) {
      try {
        File storeDirectory =
            storageManagerUtil.getTaskStoreDir(loggedStoreBaseDir, store, taskName, taskModel.getTaskMode());
        File checkpointDir = new File(storageManagerUtil.getStoreCheckpointDir(storeDirectory, checkpointId));
        FileUtils.copyDirectory(storeDirectory, checkpointDir);
      } catch (IOException exception) {
        String msg = String.format("Unable to create a copy of store directory %s into checkpoint dir %s while "
            + "attempting to recover from DeletedException", store, checkpointId);
        throw new SamzaException(msg, exception);
      }
    }
  }
Returns a single future that guarantees all the following are completed, in this order:
  1. Restore state locally by getting deleted blobs from the blob store.
  2. Create a new snapshot from restored state by backing it up on the blob store.
  3. Remove TTL from the new Snapshot and all the associated blobs in the blob store
  4. Clean up old/deleted Snapshot
  5. Create and write the new checkpoint to checkpoint topic
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerRestoreUtil.java