private CompletionStage<Void> deleteFile(FileIndex fileIndex, Metadata metadata) {
    List<CompletionStage<Void>> deleteFutures = new ArrayList<>();
    List<FileBlob> fileBlobs = fileIndex.getBlobs();
    for (FileBlob fileBlob : fileBlobs) {
      LOG.debug("Deleting file: {} blobId: {} from blob store.", fileIndex.getFileName(), fileBlob.getBlobId());
      String opName = "deleteFile: " + fileIndex.getFileName() + " blobId: " + fileBlob.getBlobId();
      Supplier<CompletionStage<Void>> fileDeletionAction = () ->
          blobStoreManager.delete(fileBlob.getBlobId(), metadata).toCompletableFuture();
      CompletableFuture<Void> fileDeletionFuture =
          FutureUtil.executeAsyncWithRetries(opName, fileDeletionAction, isCauseNonRetriable(), executor, retryPolicyConfig);
      deleteFutures.add(fileDeletionFuture);
    }

    return CompletableFuture.allOf(deleteFutures.toArray(new CompletableFuture[0]));
  }
Delete a {@link FileIndex} from the remote store by deleting all {@link FileBlob}s associated with it.
@param fileIndex FileIndex of the file to delete from the remote store.
@param metadata
@return a future that completes when the FileIndex has been marked for deletion in the remote blob store.
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java