public CompletionStage<Void> cleanSnapshotIndex(String snapshotIndexBlobId, SnapshotIndex snapshotIndex, Metadata requestMetadata) {
    DirIndex dirIndex = snapshotIndex.getDirIndex();
    CompletionStage<Void> storeDeletionFuture =
        cleanUpDir(dirIndex, requestMetadata) // delete files and sub-dirs previously marked for removal
            .thenComposeAsync(v ->
                deleteDir(dirIndex, requestMetadata), executor) // deleted files and dirs still present
            .thenComposeAsync(v -> deleteSnapshotIndexBlob(snapshotIndexBlobId, requestMetadata), executor) // delete the snapshot index blob
            .exceptionally(ex -> {
              Throwable unwrappedException = FutureUtil.unwrapExceptions(CompletionException.class,
                  FutureUtil.unwrapExceptions(SamzaException.class, ex));
              // If a blob is already deleted, do not fail -> this may happen if after we restore a
              // deleted checkpoint and then try to clean up old checkpoint.
              if (unwrappedException instanceof DeletedException) {
                LOG.warn("Request {} received DeletedException on trying to clean up SnapshotIndex {}. Ignoring the error.",
                    requestMetadata, snapshotIndexBlobId);
                return null;
              }
              String msg = String.format("Request %s received error deleting/cleaning up SnapshotIndex: %s",
                  requestMetadata, snapshotIndexBlobId);
              throw new SamzaException(msg, ex);
            });
    return storeDeletionFuture;
  }
Cleans up a SnapshotIndex by recursively deleting all blobs associated with files/subdirs inside the SnapshotIndex
and finally deletes SnapshotIndex blob itself.
@param snapshotIndexBlobId Blob if of SnapshotIndex
@param snapshotIndex SnapshotIndex to delete
@param requestMetadata Metadata of the request
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java