public CompletableFuture<SnapshotIndex> getSnapshotIndex(String blobId, Metadata metadata, boolean getDeleted) {
    Preconditions.checkState(StringUtils.isNotBlank(blobId));
    String opName = "getSnapshotIndex: " + blobId;
    return FutureUtil.executeAsyncWithRetries(opName, () -> {
      ByteArrayOutputStream indexBlobStream = new ByteArrayOutputStream(); // no need to close ByteArrayOutputStream
      return blobStoreManager.get(blobId, indexBlobStream, metadata, getDeleted).toCompletableFuture()
          .thenApplyAsync(f -> snapshotIndexSerde.fromBytes(indexBlobStream.toByteArray()), executor)
          .handle((snapshotIndex, ex) -> {
            if (ex != null) {
              throw new SamzaException(String.format("Unable to get SnapshotIndex blob. The blob ID is : %s", blobId), ex);
            }
            return snapshotIndex;
          });
    }, isCauseNonRetriable(), executor, retryPolicyConfig);
  }
GETs the {@link SnapshotIndex} from the blob store.
@param blobId blob ID of the {@link SnapshotIndex} to get
@return a Future containing the {@link SnapshotIndex}
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java