CompletableFuture<List<CloudBlobMetadata>> queryMetadataAsync(String partitionPath, SqlQuerySpec sqlQuerySpec,
      Timer timer) throws CosmosException {

    CompletableFuture<List<CloudBlobMetadata>> resultFuture = new CompletableFuture<>();
    CosmosQueryRequestOptions queryRequestOptions = new CosmosQueryRequestOptions();
    queryRequestOptions.setPartitionKey(new PartitionKey(partitionPath));
    queryRequestOptions.setResponseContinuationTokenLimitInKb(continuationTokenLimitKb);

    Timer.Context operationTime = timer.time();
    List<CloudBlobMetadata> metadataList = new ArrayList<>();

    // Since the requestCharge is being used inside lambda expression below, it needs to be atomic.
    AtomicReference<Double> requestCharge = new AtomicReference<>(0.0);

    // Execute cosmos query
    CosmosPagedFlux<CloudBlobMetadata> pagedFluxResponse = executeCosmosQueryAsync(sqlQuerySpec, queryRequestOptions);
    pagedFluxResponse.byPage().subscribe(fluxResponse -> {
      logger.debug("Got a page of query result with " + fluxResponse.getResults().size() + " items(s)"
          + " and request charge of " + fluxResponse.getRequestCharge());
      requestCharge.updateAndGet(v -> v + fluxResponse.getRequestCharge());
      metadataList.addAll(fluxResponse.getResults());
    }, throwable -> {
      Exception ex = Utils.extractFutureExceptionCause(throwable);
      operationTime.stop();
      if (ex instanceof CosmosException) {
        logger.warn("Query {} on partition {} got {}", sqlQuerySpec.getQueryText(), partitionPath,
            ((CosmosException) ex).getStatusCode());
      }
      resultFuture.completeExceptionally(ex);
    }, () -> {
      operationTime.stop();
      if (requestCharge.get() >= requestChargeThreshold) {
        logger.info("Query partition {} request charge {} for {} records", partitionPath, requestCharge,
            metadataList.size());
      }
      resultFuture.complete(metadataList);
    });
    return resultFuture;
  }
Get the list of blobs in the specified partition matching the specified DocumentDB query spec asynchronously.
@param partitionPath the partition to query.
@param sqlQuerySpec representing the sql query.
@param timer the {@link Timer} to use to record query time (excluding waiting).
@return a {@link CompletableFuture} that will eventually contain a list of {@link CloudBlobMetadata} referencing
        the matching blobs.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/CosmosDataAccessor.java