public Map<String, Pair<String, SnapshotIndex>> getStoreSnapshotIndexes(String jobName, String jobId, String taskName,
      Checkpoint checkpoint, Set<String> storesToBackupOrRestore, boolean getDeleted) {
    //TODO MED shesharma document error handling (checkpoint ver, blob not found, getBlob)
    if (checkpoint == null) {
      LOG.debug("No previous checkpoint found for taskName: {}", taskName);
      return ImmutableMap.of();
    }

    if (checkpoint.getVersion() == 1) {
      LOG.warn("Checkpoint version 1 is not supported for blob store backup and restore.");
      return ImmutableMap.of();
    }

    Map<String, CompletableFuture<Pair<String, SnapshotIndex>>>
        storeSnapshotIndexFutures = new HashMap<>();

    CheckpointV2 checkpointV2 = (CheckpointV2) checkpoint;
    Map<String, Map<String, String>> factoryToStoreSCMs = checkpointV2.getStateCheckpointMarkers();
    Map<String, String> storeSnapshotIndexBlobIds = factoryToStoreSCMs.get(BlobStoreStateBackendFactory.class.getName());

    if (storeSnapshotIndexBlobIds != null) {
      storeSnapshotIndexBlobIds.forEach((storeName, snapshotIndexBlobId) -> {
        if (storesToBackupOrRestore.contains(storeName)) {
          try {
            LOG.debug("Getting snapshot index for taskName: {} store: {} blobId: {} with getDeleted set to {}",
                taskName, storeName, snapshotIndexBlobId, getDeleted);
            Metadata requestMetadata =
                new Metadata(Metadata.SNAPSHOT_INDEX_PAYLOAD_PATH, Optional.empty(), jobName, jobId, taskName, storeName);
            CompletableFuture<SnapshotIndex> snapshotIndexFuture =
                getSnapshotIndex(snapshotIndexBlobId, requestMetadata, getDeleted).toCompletableFuture();
            Pair<CompletableFuture<String>, CompletableFuture<SnapshotIndex>> pairOfFutures =
                Pair.of(CompletableFuture.completedFuture(snapshotIndexBlobId), snapshotIndexFuture);

            // save the future and block once in the end instead of blocking for each request.
            storeSnapshotIndexFutures.put(storeName, FutureUtil.toFutureOfPair(pairOfFutures));
          } catch (Exception e) {
            throw new SamzaException(
                String.format("Error getting SnapshotIndex for blobId: %s for taskName: %s store: %s",
                    snapshotIndexBlobId, taskName, storeName), e);
          }
        } else {
          LOG.debug("SnapshotIndex blob id {} for store {} is not present in the set of stores to be backed up/restores: {}",
              snapshotIndexBlobId, storeName, storesToBackupOrRestore);
        }
      });
    } else {
      LOG.debug("No store SCMs found for blob store state backend in for taskName: {} in checkpoint {}",
          taskName, checkpointV2.getCheckpointId());
    }

    try {
      return FutureUtil.toFutureOfMap(storeSnapshotIndexFutures).join();
    } catch (Exception e) {
      throw new SamzaException(
          String.format("Error while waiting to get store snapshot indexes for task %s", taskName), e);
    }
  }
Get the blob id of {@link SnapshotIndex} and {@link SnapshotIndex}es for the provided {@code task}
in the provided {@code checkpoint}.
@param jobName job name is used to build request metadata
@param jobId job id is used to build request metadata
@param taskName task name to get the store state checkpoint markers and snapshot indexes for
@param checkpoint {@link Checkpoint} instance to get the store state checkpoint markers from. Only
                  {@link CheckpointV2} and newer are supported for blob stores.
@param storesToBackupOrRestore set of store names to be backed up or restored
@param getDeleted tries gets a deleted but not yet compacted SnapshotIndex from the blob store.
@return Map of store name to its blob id of snapshot indices and their corresponding snapshot indices for the task.
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java