@Override
    public Integer call() throws Exception {
      /*
        Jobs wait for execution until workers become available.
        If 100 jobs are submitted to 5 workers, only 5 jobs run while the rest wait.
        Partitions can be reassigned while jobs are in the queue.
        Verify ownership of the partition before job initiation.
        For backward compatibility with previous impl of cloudDestination
       */
      if (!isPartitionOwned(partitionId)) {
        logger.info("[COMPACT] Skipping compaction partition-{} as it is disowned", partitionId.toPathString());
        return 0;
      }

      // For backward compatibility with previous impl of cloudDestination
      if (cloudDestination.isCompactionStopped(partitionId.toPathString(), compactor)) {
        logger.info("[COMPACT] Skipping compaction partition-{} due to shutdowm", partitionId.toPathString());
        return 0;
      }

      try {
        /*
          Atomically increment the num blobs erased. Do not rely on the higher loop to collect all Futures.
          A single slow worker can prevent the higher loop from gathering results from completed Futures.
          If the higher loop is interrupted, then it throws an exc and we eventually return 0 as num blobs erased,
          which is not accurate. However, return the result for backward compatibility.
         */
        return numBlobsErased.addAndGet(cloudDestination.compactPartition(partitionId.toPathString(), compactor));
      } catch (Throwable throwable) {
        // Swallow all exceptions
        azureMetrics.compactionTaskErrorCount.inc();
        logger.error("[COMPACT] Failed to compact partition-{} due to {}",
            partitionId.toPathString(), throwable);
      }
      return 0;
    }
@return whether the compactor is shutting down.
ambry-cloud/src/main/java/com/github/ambry/cloud/CloudStorageCompactor.java