void resumeCompaction(byte[] bundleReadBuffer) throws StoreException {
    this.bundleReadBuffer = bundleReadBuffer;
    if (srcIndex == null) {
      throw new IllegalStateException("Compactor has not been initialized");
    } else if (compactionLog == null) {
      throw new IllegalStateException("There is no compaction to resume");
    }
    /*
    A single compaction job could be performed across multiple compaction cycles (if there aren't enough swap spaces to
    complete the job in one cycle). Therefore, we loop and perform PREPARE, COPY, COMMIT and CLEANUP until the
    CompactionLog reports the job as DONE.
    Short description of each of the phases:
    1. PREPARE - Initial state of a compaction cycle. Signifies that the cycle has just started.
    2. COPY - Valid data from the segments under compaction is copied over to the swap spaces. If all the
    data cannot be copied, it auto splits the compaction job into two cycles.
    3. COMMIT - Adds the new log segments to the application log and atomically updates the set of index segments
    4. CLEANUP - Cleans up the old log segments and their index files
     */

    logger.trace("resumeCompaction() started for {}", storeId);
    runningLatch = new CountDownLatch(1);
    compactionInProgress.set(true);
    if (config.storeContainerDeletionEnabled) {
      getDeprecatedContainers();
      logger.info("Deprecated containers are {} for {}", deprecatedContainers, storeId);
    }
    long startTime = time.milliseconds();
    try {
      while (isActive && !compactionLog.getCompactionPhase().equals(CompactionLog.Phase.DONE)) {
        CompactionLog.Phase phase = compactionLog.getCompactionPhase();
        logger.debug("Starting cycle to compact {} in {}", compactionLog.getCompactionDetails(), storeId);
        switch (phase) {
          case PREPARE:
            logger.debug("Compaction PREPARE started for {} in {}", compactionLog.getCompactionDetails(), storeId);
            compactionLog.markCopyStart();
            // fall through to COPY
          case COPY:
            logger.debug("Compaction COPY started for {} in {}", compactionLog.getCompactionDetails(), storeId);
            copy();
            if (isActive) {
              compactionLog.markCommitStart();
              logger.debug("Compaction COMMIT started for {} in {}", compactionLog.getCompactionDetails(), storeId);
              commit(false);
              compactionLog.markCleanupStart();
              logger.debug("Compaction CLEANUP started for {} in {}", compactionLog.getCompactionDetails(), storeId);
              waitTillRefCountZero();
              logger.debug("Wait for ref count to go to 0 has been finished {} in {}",
                  compactionLog.getCompactionDetails(), storeId);
              cleanup(false);
              logger.debug("Completing for {} in {}", compactionLog.getCompactionDetails(), storeId);
              compactionLog.markCycleComplete();
            }
            break;
          default:
            throw new IllegalStateException("Illegal compaction phase: " + phase);
        }
      }
      endCompaction();
    } catch (InterruptedException | IOException e) {
      if (e instanceof IOException) {
        diskMetrics.diskCompactionErrorDueToDiskFailureCount.inc();
        logger.error("Compaction of store {} failed due to disk issue.", dataDir);
      }
      throw new StoreException("Exception during compaction", e, StoreErrorCodes.UnknownError);
    } finally {
      compactionInProgress.set(false);
      runningLatch.countDown();
      logger.trace("resumeCompaction() ended for {}", storeId);

      if (currentCompactionDetails.get() != null) {
        // currentCompactionDetails could be null in test context
        if (currentCompactionDetails.get().isFullRange()) {
          // Measure time for full range compaction
          srcMetrics.fullScanCompactionFinishedForStoreTimeInMs.update(
              time.milliseconds() - startTime, TimeUnit.MILLISECONDS);
        } else {
          // Measure time for partial range compaction
          srcMetrics.partialRangeCompactionFinishedForStoreTimeInMs.update(
              time.milliseconds() - startTime, TimeUnit.MILLISECONDS);
        }
      }
    }
    this.bundleReadBuffer = null;
  }
Resumes compaction from where it was left off.
@throws IllegalStateException if the compactor has not been initialized or if there is no compaction to resume.
@throws StoreException if there were exceptions reading to writing to store components.
@param bundleReadBuffer the preAllocated buffer for bundle read in compaction copy phase.
ambry-store/src/main/java/com/github/ambry/store/BlobStoreCompactor.java