public CompletableFuture<String> putSnapshotIndex(SnapshotIndex snapshotIndex) {
    byte[] bytes = snapshotIndexSerde.toBytes(snapshotIndex);
    String opName = "putSnapshotIndex for checkpointId: " + snapshotIndex.getSnapshotMetadata().getCheckpointId();
    return FutureUtil.executeAsyncWithRetries(opName, () -> {
      InputStream inputStream = new ByteArrayInputStream(bytes); // no need to close ByteArrayInputStream
      SnapshotMetadata snapshotMetadata = snapshotIndex.getSnapshotMetadata();
      Metadata metadata = new Metadata(Metadata.SNAPSHOT_INDEX_PAYLOAD_PATH, Optional.of((long) bytes.length),
          snapshotMetadata.getJobName(), snapshotMetadata.getJobId(), snapshotMetadata.getTaskName(),
          snapshotMetadata.getStoreName());
      return blobStoreManager.put(inputStream, metadata).toCompletableFuture();
    }, isCauseNonRetriable(), executor, retryPolicyConfig);
  }
PUTs the {@link SnapshotIndex} to the blob store.
@param snapshotIndex SnapshotIndex to put.
@return a Future containing the blob ID of the {@link SnapshotIndex}.
samza-core/src/main/java/org/apache/samza/storage/blobstore/util/BlobStoreUtil.java