CompletableFuture<CloudBlobMetadata> updateMetadataAsync(BlobId blobId, Map<String, String> updateFields) {

    CompletableFuture<CloudBlobMetadata> resultFuture = new CompletableFuture<>();
    Timer.Context documentReadTimer = azureMetrics.documentReadTime.time();
    AtomicReference<Timer.Context> documentUpdateTimer = new AtomicReference<>();

    // Read the existing record
    cosmosAsyncContainer.readItem(blobId.getID(), new PartitionKey(blobId.getPartition().toPathString()),
        CloudBlobMetadata.class).toFuture().handle((readItemResponse, throwableOnRead) -> {
      // Stop the document read timer
      documentReadTimer.stop();
      if (throwableOnRead != null) {
        // If there is an error in reading the current metadata record, complete the result exceptionally.
        resultFuture.completeExceptionally(throwableOnRead);
        return null;
      } else {
        // Get the metadata from cosmos response
        CloudBlobMetadata receivedMetadata = readItemResponse.getItem();

        // Update the metadata record only if value has changed
        Map<String, String> receivedFields = receivedMetadata.toMap();
        Map<String, String> fieldsToUpdate = updateFields.entrySet()
            .stream()
            .filter(map -> !String.valueOf(updateFields.get(map.getKey())).equals(receivedFields.get(map.getKey())))
            .collect(Collectors.toMap(Map.Entry::getKey, map -> String.valueOf(map.getValue())));
        if (fieldsToUpdate.isEmpty()) {
          logger.debug("No change in value for {} in blob {}", updateFields.keySet(), blobId.getID());
          // If there is no change in value of metadata, complete the result.
          resultFuture.complete(null);
        }

        // For testing conflict handling
        if (updateCallback != null) {
          try {
            updateCallback.call();
          } catch (Exception ex) {
            logger.error("Error in update callback", ex);
          }
        }

        // Update the modified fields
        fieldsToUpdate.forEach(receivedFields::put);

        // Set condition to ensure we don't clobber a concurrent update
        CosmosItemRequestOptions requestOptions = new CosmosItemRequestOptions();
        requestOptions.setIfMatchETag(readItemResponse.getETag());

        // Replace the item in cosmos db asynchronously.
        documentUpdateTimer.set(azureMetrics.documentUpdateTime.time());
        return cosmosAsyncContainer.replaceItem(CloudBlobMetadata.fromMap(receivedFields), blobId.getID(),
            new PartitionKey(blobId.getPartition().toPathString()), requestOptions).toFuture();
      }
    }).thenCompose(result -> {
      // Since previous stage returns CompletionStage<CompletionStage<T>>, using #thenCompose helps to flatten the result.
      return result;
    }).whenComplete((replaceItemResponse, throwable) -> {
      if (documentUpdateTimer.get() != null) {
        documentUpdateTimer.get().stop();
      }
      if (throwable != null) {
        Exception ex = Utils.extractFutureExceptionCause(throwable);
        if (((CosmosException) ex).getStatusCode() == HttpConstants.StatusCodes.PRECONDITION_FAILED) {
          // Received an exception during metadata replacement due to concurrent update. Increase the conflict counter
          azureMetrics.blobUpdateConflictCount.inc();
        }
        resultFuture.completeExceptionally(ex);
      } else {
        // Updated the metadata record successfully
        resultFuture.complete(replaceItemResponse.getItem());
      }
    });

    return resultFuture;
  }
Update the blob metadata document in the CosmosDB collection asynchronously.
@param blobId the {@link BlobId} for which metadata is replaced.
@param updateFields Map of field names and new values to update.
@return a {@link CompletableFuture} that will eventually contain the {@link CloudBlobMetadata} returned by the
        operation, if successful or {@code Null} if the field already has the specified value. If the record is not
        found or an error occurred, it will contain a {@link CosmosException}.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/CosmosDataAccessor.java