private static CompletableFuture<Map<String, String>> backupRecoveredStore(JobContext jobContext,
      ContainerModel containerModel, Config config, TaskName taskName, Set<String> storesToBackup,
      CheckpointId newCheckpointId, File loggedStoreBaseDirectory, BlobStoreManager blobStoreManager,
      MetricsRegistry metricsRegistry, ExecutorService executor) {

    BlobStoreBackupManagerMetrics blobStoreBackupManagerMetrics = new BlobStoreBackupManagerMetrics(metricsRegistry);
    BlobStoreBackupManager blobStoreBackupManager =
        new BlobStoreBackupManager(jobContext.getJobModel(), containerModel, containerModel.getTasks().get(taskName),
            executor, blobStoreBackupManagerMetrics, config, SystemClock.instance(), loggedStoreBaseDirectory,
            new StorageManagerUtil(), blobStoreManager);

    // create checkpoint dir as a copy of store dir
    createCheckpointDirFromStoreDirCopy(taskName, containerModel.getTasks().get(taskName),
        loggedStoreBaseDirectory, storesToBackup, newCheckpointId);
    // upload to blob store and return future
    return blobStoreBackupManager.upload(newCheckpointId, new HashMap<>());
  }
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