@Override
    public void run() {
      isRunning = true;
      try {
        logger.info("Starting compaction thread for {}", mountPath);
        // complete any compactions in progress
        for (BlobStore store : stores) {
          logger.trace("{} being checked for resume", store);
          if (store.isStarted()) {
            logger.trace("{} is started and eligible for resume check", store);
            CompactionDetails details = null;
            boolean isFullRange = false;
            try {
              details = store.getCompactionDetailsInProgress();
              if (details != null) {
                isFullRange = details.isFullRange();
              }
              metrics.markCompactionStart(false, isFullRange);
              store.maybeResumeCompaction(bundleReadBuffer);
            } catch (Exception e) {
              metrics.compactionErrorCount.inc();
              logger.error("Compaction of store {} failed on resume. Continuing with the next store", store, e);
              storesToSkip.add(store);
            } finally {
              metrics.markCompactionStop(isFullRange);
            }
          }
        }
        // continue to do compactions as required.
        long expectedNextCheckTime = time.milliseconds() + waitTimeMs;
        if (triggers.contains(Trigger.PERIODIC)) {
          storesToCheck.addAll(stores);
        }
        int storesNoCompaction = 0;
        while (enabled) {
          try {
            storesNoCompaction = 0;
            logger.info("[CAPACITY] Going to check compaction on {}", mountPath);
            while (enabled && storesToCheck.peek() != null) {
              BlobStore store = storesToCheck.poll();
              logger.trace("{} being checked for compaction", store);
              boolean compactionStarted = false;
              CompactionDetails details = null;
              try {
                if (store.isStarted() && !storesToSkip.contains(store) && !storesDisabledCompaction.contains(store)) {
                  logger.info("[CAPACITY] {} is started and is being checked for compaction eligibility", store);
                  details = getCompactionDetails(store);
                  if (details != null) {
                    logger.trace("Generated {} as details for {}", details, store);
                    metrics.markCompactionStart(true, details.isFullRange());
                    compactionStarted = true;
                    store.compact(details, bundleReadBuffer);
                  } else {
                    storesNoCompaction++;
                    logger.info("[CAPACITY] {} is not eligible for compaction due to empty compaction details", store);
                  }
                  if (storeConfig.storeAutoCloseLastLogSegmentEnabled && store.getReplicaId().isSealed()) {
                    store.closeLastLogSegmentIfQualified();
                  }
                }
              } catch (Exception e) {
                metrics.compactionErrorCount.inc();
                logger.error("[CAPACITY] Compaction of store {} failed. Continuing with the next store", store, e);
                storesToSkip.add(store);
              } finally {
                if (compactionStarted) {
                  metrics.markCompactionStop(details.isFullRange());
                }
              }
            }
            lock.lock();
            try {
              if (enabled) {
                if (storesToCheck.peek() == null) {
                  if (triggers.contains(Trigger.PERIODIC)) {
                    // If compaction is not run on some of stores, wait time should be reduced.
                    // The max time to reduce is waitTimeMs / 2;
                    long compensation =
                        storesNoCompaction != 0 && stores.size() > 0 ? waitTimeMs / 2 * storesNoCompaction
                            / stores.size() : 0;
                    long actualWaitTimeMs = expectedNextCheckTime - compensation - time.milliseconds();
                    logger.trace("[CAPACITY] Going to wait for {} ms in compaction thread at {}", actualWaitTimeMs,
                        mountPath);
                    time.await(waitCondition, actualWaitTimeMs);
                  } else {
                    logger.trace("[CAPACITY] Going to wait until compaction thread at {} is woken up", mountPath);
                    waitCondition.await();
                  }
                }
                if (triggers.contains(Trigger.PERIODIC) && time.milliseconds() >= expectedNextCheckTime) {
                  expectedNextCheckTime = time.milliseconds() + waitTimeMs;
                  storesToCheck.addAll(stores);
                }
              }
            } finally {
              lock.unlock();
            }
          } catch (Exception e) {
            metrics.compactionExecutorErrorCount.inc();
            logger.error("Compaction executor for {} encountered an error. Continuing", mountPath, e);
          }
        }
      } finally {
        isRunning = false;
        logger.info("Stopping compaction thread for {}", mountPath);
      }
    }
Starts by resuming any compactions that were left halfway. In steady state, it cycles through the stores at a
configurable frequency and runs compaction as required.
ambry-store/src/main/java/com/github/ambry/store/CompactionManager.java