@VisibleForTesting
  static void deleteUnusedStoresFromBlobStore(String jobName, String jobId, String taskName, StorageConfig storageConfig,
      BlobStoreConfig blobStoreConfig, Map<String, Pair<String, SnapshotIndex>> initialStoreSnapshotIndexes,
      BlobStoreUtil blobStoreUtil, ExecutorService executor) {

    List<String> storesToBackup =
        storageConfig.getStoresWithBackupFactory(BlobStoreStateBackendFactory.class.getName());
    List<String> storesToRestore =
        storageConfig.getStoresWithRestoreFactory(BlobStoreStateBackendFactory.class.getName());

    List<CompletionStage<Void>> storeDeletionFutures = new ArrayList<>();
    initialStoreSnapshotIndexes.forEach((storeName, scmAndSnapshotIndex) -> {
      if (!storesToBackup.contains(storeName) && !storesToRestore.contains(storeName)) {
        LOG.info("Removing task: {} store: {} from blob store. It is either no longer used, " +
            "or is no longer configured to be backed up or restored with blob store.", taskName, storeName);
        Metadata requestMetadata =
            new Metadata(Metadata.SNAPSHOT_INDEX_PAYLOAD_PATH, Optional.empty(), jobName, jobId, taskName, storeName);
        storeDeletionFutures.add(blobStoreUtil.cleanSnapshotIndex(scmAndSnapshotIndex.getLeft(), scmAndSnapshotIndex.getRight(), requestMetadata));
      }
    });

    FutureUtil.allOf(storeDeletionFutures).join();
  }
Deletes blob store contents for stores that were present in the last checkpoint but are either no longer
present in job configs (removed by user since last deployment) or are no longer configured to be backed
up using blob stores.

This method blocks until all the necessary store contents and snapshot index blobs have been marked for deletion.
samza-core/src/main/java/org/apache/samza/storage/blobstore/BlobStoreRestoreManager.java