private int compactPartitionBucketed(String partitionPath, String fieldName, long queryStartTime, long queryEndTime)
      throws CloudStorageException {

    if (queryEndTime - queryStartTime > TimeUnit.DAYS.toMillis(queryBucketDays)) {
      throw new IllegalArgumentException("Time window is longer than " + queryBucketDays + " days");
    }

    int totalPurged = 0;
    long progressTime = 0;
    while (!isShuttingDown()) {

      final long newQueryStartTime = queryStartTime; // just to use in lambda
      List<CloudBlobMetadata> deadBlobs = requestAgent.doWithRetries(
          () -> getDeadBlobs(partitionPath, fieldName, newQueryStartTime, queryEndTime, queryLimit), "GetDeadBlobs",
          partitionPath);
      if (deadBlobs.isEmpty()) {
        // If query returned nothing, we can mark progress up to queryEndTime
        progressTime = queryEndTime;
        break;
      }
      if (isShuttingDown()) {
        break;
      }
      totalPurged += requestAgent.doWithRetries(
          () -> AzureCompactionUtil.purgeBlobs(deadBlobs, azureBlobDataAccessor, azureMetrics, cosmosDataAccessor),
          "PurgeBlobs", partitionPath);
      vcrMetrics.blobCompactionRate.mark(deadBlobs.size());

      // Adjust startTime for next query
      CloudBlobMetadata lastBlob = deadBlobs.get(deadBlobs.size() - 1);
      progressTime = fieldName.equals(CloudBlobMetadata.FIELD_DELETION_TIME) ? lastBlob.getDeletionTime()
          : lastBlob.getExpirationTime();
      queryStartTime = progressTime;

      if (deadBlobs.size() < queryLimit) {
        // No more dead blobs to query.
        break;
      }

      if (totalPurged >= purgeLimit) {
        // Reached the purge threshold, give other partitions a chance.
        break;
      }
    }

    if (progressTime > 0) {
      updateCompactionProgress(partitionPath, fieldName, progressTime);
    }
    return totalPurged;
  }
Purge the inactive blobs in the specified partition.
@param partitionPath the partition to compact.
@param fieldName the field name to query on. Allowed values are {@link CloudBlobMetadata#FIELD_DELETION_TIME}
                     and {@link CloudBlobMetadata#FIELD_EXPIRATION_TIME}.
@param queryStartTime the initial query start time, which will be adjusted as compaction progresses.
@param queryEndTime the query end time.
@return the number of blobs purged or found.
@throws CloudStorageException if the compaction fails.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/AzureStorageCompactor.java