private CompletableFuture<UpdateResponse> updateBlobMetadata(BlobId blobId, Map<String, Object> updateFields,
      CloudUpdateValidator cloudUpdateValidator) {
    Objects.requireNonNull(blobId, "BlobId cannot be null");
    updateFields.keySet().forEach(field -> Objects.requireNonNull(updateFields.get(field)));

    CompletableFuture<UpdateResponse> resultFuture = new CompletableFuture<>();

    // We update the blob metadata value in two places:
    // 1) the blob storage entry metadata (so GET's can be served entirely from ABS)
    // 2) the CosmosDB metadata collection
    azureBlobDataAccessor.updateBlobMetadataAsync(blobId, updateFields, cloudUpdateValidator)
        .whenComplete((blobStorageUpdateResponse, throwableOnBlobStorageUpdate) -> {
          if (throwableOnBlobStorageUpdate != null) {
            // There is an error updating the blob metadata in ABS.
            azureMetrics.blobUpdateErrorCount.inc();
            Exception blobUpdateException = Utils.extractFutureExceptionCause(throwableOnBlobStorageUpdate);
            if (blobUpdateException instanceof BlobStorageException
                && ((BlobStorageException) blobUpdateException).getErrorCode() == BlobErrorCode.BLOB_NOT_FOUND) {
              // There is a corner case where compaction of this partition hit partial failure leaving the record
              // in Cosmos but not ABS.  If that happens, a late arriving update event can get stuck in a loop
              // where findMissingKeys says the blob exists (because Cosmos has it), but the subsequent update
              // attempt fails.  So we check for that case here.
              cosmosDataAccessor.getMetadataOrNullAsync(blobId).whenComplete((cosmosMetadata, throwableOnCosmosGet) -> {
                if (throwableOnCosmosGet != null) {
                  // Received an exception when trying to find blob metadata in cosmos. Complete the result future with the original exception from ABS.
                  resultFuture.completeExceptionally(
                      toCloudStorageException("Error updating blob metadata: " + blobId, blobUpdateException));
                } else {
                  if (cosmosMetadata != null) {
                    // If the blob is found in cosmos. Check if it is a compaction candidate. If yes, delete the record in the cosmos.
                    if (cosmosMetadata.isCompactionCandidate(
                        TimeUnit.HOURS.toMillis(cloudConfig.cloudBlobCompactionIntervalHours))) {
                      logger.warn("Inconsistency: Cosmos contains record for inactive blob {}, removing it.",
                          blobId.getID());
                      cosmosDataAccessor.deleteMetadataAsync(cosmosMetadata)
                          .whenComplete((deleteMetadataResponse, throwableOnCosmosDelete) -> {
                            // Blob is deleted in cosmos too now.
                            azureMetrics.blobUpdateRecoverCount.inc();
                            resultFuture.completeExceptionally(
                                toCloudStorageException("Error updating blob metadata: " + blobId,
                                    blobUpdateException));
                          });
                    } else {
                      // If the blob is still active but ABS does not have it, we are in deeper trouble.
                      logger.error("Inconsistency: Cosmos contains record for active blob {} that is missing from ABS!",
                          blobId.getID());
                      resultFuture.completeExceptionally(
                          toCloudStorageException("Error updating blob metadata: " + blobId, blobUpdateException));
                    }
                  } else {
                    // Blob is not found in cosmos too. Complete the result future.
                    resultFuture.completeExceptionally(
                        toCloudStorageException("Error updating blob metadata: " + blobId, blobUpdateException));
                  }
                }
              });
            } else {
              // If the exception from ABS is any thing other than NOT_FOUND, complete the result future immediately.
              CloudStorageException cse =
                  toCloudStorageException("Error updating blob metadata: " + blobId, blobUpdateException);
              resultFuture.completeExceptionally(cse);
            }
          } else {
            // Blob is updated in blob storage. Now update in cosmos.
            // Note: even if nothing changed in blob storage, still attempt to update Cosmos since this could be a retry
            // of a request where ABS was updated but Cosmos update failed.
            AtomicReference<Map<String, String>> metadataMap =
                new AtomicReference<>(blobStorageUpdateResponse.metadata);
            AtomicBoolean updatedStorage = new AtomicBoolean(blobStorageUpdateResponse.wasUpdated);
            cosmosDataAccessor.updateMetadataAsync(blobId, metadataMap.get())
                .whenComplete((cosmosMetadata, throwableOnCosmosUpdate) -> {
                  AtomicBoolean updatedCosmos = new AtomicBoolean(false);
                  if (throwableOnCosmosUpdate != null) {
                    // Received an exception when trying to update the metadata in cosmos.
                    Exception ex = Utils.extractFutureExceptionCause(throwableOnCosmosUpdate);
                    if (ex instanceof CosmosException && ((CosmosException) ex).getStatusCode() == NOTFOUND) {
                      // If the exception is due to blob not being found in Cosmos, it is an inconsistent state since
                      //blob exists in ABS. Recover by inserting the updated map into cosmos.
                      azureMetrics.blobUpdateRecoverCount.inc();
                      cosmosDataAccessor.upsertMetadataAsync(CloudBlobMetadata.fromMap(metadataMap.get()))
                          .whenComplete((updatedCosmosMetadata, throwableOnCosmosUpsert) -> {
                            if (throwableOnCosmosUpsert != null) {
                              azureMetrics.blobUpdateErrorCount.inc();
                              resultFuture.completeExceptionally(
                                  toCloudStorageException("Error updating blob metadata: " + blobId,
                                      (Exception) throwableOnCosmosUpsert));
                            } else {
                              updatedCosmos.set(updatedCosmosMetadata != null);
                              resultFuture.complete(new UpdateResponse(updatedStorage.get() || updatedCosmos.get(),
                                  blobStorageUpdateResponse.metadata));
                            }
                          });
                    } else {
                      azureMetrics.blobUpdateErrorCount.inc();
                      // We got some other exception apart from blob_not_found when updating cosmos. Consider the update operation as failed.
                      resultFuture.completeExceptionally(
                          toCloudStorageException("Error updating blob metadata: " + blobId, ex));
                    }
                  } else {
                    // Updated cosmos successfully.
                    updatedCosmos.set(cosmosMetadata != null);
                    azureMetrics.blobUpdatedCount.inc();
                    logger.debug("Updated metadata response {} for blob {} metadata fields {} to values {}.",
                        updatedStorage.get() || updatedCosmos.get(), blobId, updateFields.keySet(),
                        updateFields.values());
                    resultFuture.complete(new UpdateResponse(updatedStorage.get() || updatedCosmos.get(),
                        blobStorageUpdateResponse.metadata));
                  }
                });
          }
        });
    return resultFuture;
  }
Update the metadata for the specified blob asynchronously.
@param blobId The {@link BlobId} to update.
@param updateFields map of fields and new values to update.
@param cloudUpdateValidator {@link CloudUpdateValidator} passed by the caller to validate the update.
@return a {@link CompletableFuture} that will eventually contain either the {@link UpdateResponse} object
        containing updated metadata or an exception if an error occurred.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/AzureCloudDestination.java