public CompletableFuture<UpdateResponse> updateBlobMetadataAsync(BlobId blobId, Map<String, Object> updateFields,
      CloudUpdateValidator cloudUpdateValidator) {

    CompletableFuture<UpdateResponse> resultFuture = new CompletableFuture<>();
    Objects.requireNonNull(blobId, "BlobId cannot be null");
    updateFields.keySet()
        .forEach(field -> Objects.requireNonNull(updateFields.get(field), String.format("%s cannot be null", field)));

    Timer.Context storageTimer = azureMetrics.blobUpdateTime.time();
    AtomicReference<Map<String, String>> metadata = new AtomicReference<>();

    // TODO: Pass executor instead of default async method.
    FutureUtils.orTimeout(storageClient.getPropertiesWithResponse(blobId, defaultRequestConditions), requestTimeout)
        .thenCompose((blobProperties) -> {
          // Got the current metadata record of the blob. Check for validity of the update and update the blob metadata.
          String etag = blobProperties.getETag();
          metadata.set(blobProperties.getMetadata());

          try {
            // Validate the sanity of update operation by comparing the fields we want to update against existing metadata
            // fields in Azure. This can throw StoreExceptions which are propagated to caller.
            if (!cloudUpdateValidator.validateUpdate(CloudBlobMetadata.fromMap(metadata.get()), blobId, updateFields)) {
              storageTimer.stop();
              resultFuture.complete(new UpdateResponse(false, metadata.get()));
            }
          } catch (StoreException e) {
            // Received an exception during the update validation. Complete the response future.
            storageTimer.stop();
            throw new CompletionException(e);
          }

          if (!resultFuture.isDone()) {
            // Check for list of changed fields
            Map<String, String> changedFields = updateFields.entrySet()
                .stream()
                .filter(entry -> !String.valueOf(entry.getValue()).equals(metadata.get().get(entry.getKey())))
                .collect(Collectors.toMap(Map.Entry::getKey, entry -> String.valueOf(entry.getValue())));

            // Update the record only if any of its values have changed
            if (changedFields.size() <= 0) {
              // There are no changed fields to update
              storageTimer.stop();
              resultFuture.complete(new UpdateResponse(false, metadata.get()));
            } else {
              // Modify the metadata with changes
              changedFields.forEach(metadata.get()::put);

              // Invoke the update Callback. This is only needed for our tests.
              if (updateCallback != null) {
                try {
                  updateCallback.call();
                } catch (Exception ex) {
                  logger.error("Error in update callback", ex);
                }
              }

              // Set condition to ensure we don't clobber a concurrent update
              BlobRequestConditions blobRequestConditions = new BlobRequestConditions().setIfMatch(etag);

              // Update the metadata
              return FutureUtils.orTimeout(
                  storageClient.setMetadataWithResponse(blobId, metadata.get(), blobRequestConditions, Context.NONE),
                  requestTimeout);
            }
          }
          return null;
        })
        .whenComplete((response, throwable) -> {
          storageTimer.stop();
          if (throwable != null) {
            Exception ex = Utils.extractFutureExceptionCause(throwable);
            if (ex instanceof BlobStorageException) {
              if (isNotFoundError(((BlobStorageException) ex).getErrorCode())) {
                logger.warn("Blob {} not found, cannot update {}.", blobId, updateFields.keySet());
              } else if (((BlobStorageException) ex).getErrorCode() == BlobErrorCode.CONDITION_NOT_MET) {
                logger.warn("Blob {} update condition not met, cannot update {}.", blobId, updateFields.keySet());
                azureMetrics.blobUpdateConflictCount.inc();
              }
            }
            resultFuture.completeExceptionally(ex);
          } else {
            resultFuture.complete(new UpdateResponse(true, metadata.get()));
          }
        });
    return resultFuture;
  }
Update the metadata for the specified blob asynchronously.
@param blobId The {@link BlobId} to update.
@param updateFields Map of field names and new values to modify.
@param cloudUpdateValidator {@link CloudUpdateValidator} validator for the update passed by the caller.
@return a CompletableFuture that will eventually contain the {@link UpdateResponse} with the updated metadata.
        If there was an error on ABS side, it will complete exceptionally containing the BlobStorageException.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/AzureBlobDataAccessor.java