CompletableFuture<List<CloudBlobMetadata>> getDeadBlobsAsync(String partitionPath, String fieldName, long startTime,
      long endTime, int maxEntries) {

    CompletableFuture<List<CloudBlobMetadata>> resultFuture = new CompletableFuture<>();
    String deadBlobsQuery;
    if (fieldName.equals(CloudBlobMetadata.FIELD_DELETION_TIME)) {
      deadBlobsQuery = DELETED_BLOBS_QUERY;
    } else if (fieldName.equals(CloudBlobMetadata.FIELD_EXPIRATION_TIME)) {
      deadBlobsQuery = EXPIRED_BLOBS_QUERY;
    } else {
      throw new IllegalArgumentException("Invalid field: " + fieldName);
    }

    SqlQuerySpec sqlQuerySpec = new SqlQuerySpec(deadBlobsQuery, new SqlParameter(LIMIT_PARAM, maxEntries),
        new SqlParameter(START_TIME_PARAM, startTime), new SqlParameter(END_TIME_PARAM, endTime));
    CosmosQueryRequestOptions queryRequestOptions = new CosmosQueryRequestOptions();
    queryRequestOptions.setPartitionKey(new PartitionKey(partitionPath));
    queryRequestOptions.setResponseContinuationTokenLimitInKb(continuationTokenLimitKb);

    List<CloudBlobMetadata> deadBlobsList = new ArrayList<>();
    AtomicReference<Double> requestCharge = new AtomicReference<>(0.0);

    Timer.Context deadBlobsQueryTime = azureMetrics.deadBlobsQueryTime.time();

    // Execute cosmos query
    CosmosPagedFlux<CloudBlobMetadata> pagedFluxResponse = executeCosmosQueryAsync(sqlQuerySpec, queryRequestOptions);
    //Set the maximum entries to be as returned in each page. There seems to be no way to set maximum number of entries across all pages.
    pagedFluxResponse.byPage(maxEntries).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());
      deadBlobsList.addAll(fluxResponse.getResults());
    }, throwable -> {
      deadBlobsQueryTime.stop();
      if (throwable instanceof CosmosException) {
        logger.warn("Dead blobs query {} partition {} got {}", deadBlobsQuery, partitionPath,
            ((CosmosException) throwable).getStatusCode());
      }
      resultFuture.completeExceptionally(throwable);
    }, () -> {
      deadBlobsQueryTime.stop();
      if (requestCharge.get() >= requestChargeThreshold) {
        logger.info("Dead blobs query partition {} endTime {} request charge {} for {} records", partitionPath,
            new Date(endTime), requestCharge.get(), deadBlobsList.size());
      }
      resultFuture.complete(deadBlobsList);
    });
    return resultFuture;
  }
Get the list of blobs in the specified partition that have been deleted or expired for at least the
configured retention period asynchronously.
@param partitionPath the partition to query.
@param fieldName the field name to query on. Allowed values are {@link CloudBlobMetadata#FIELD_DELETION_TIME} and
                 {@link CloudBlobMetadata#FIELD_EXPIRATION_TIME}.
@param startTime the start of the query time range.
@param endTime the end of the query time range.
@param maxEntries the max number of metadata records to return.
@return a {@link CompletableFuture} that will eventually contain a list of {@link CloudBlobMetadata} referencing
        the dead blobs found or an exception if an error occurred.
@throws CosmosException if the operation failed.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/CosmosDataAccessor.java