private void compactWithRecoveryAndVerify(Log log, DiskIOScheduler diskIOScheduler, PersistentIndex index,
      List<LogSegmentName> segmentsUnderCompaction, long deleteReferenceTimeMs, boolean changeExpected,
      boolean checkSavedBytesReported) throws Exception {
    // getting these from the "real" log and indexes
    long logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
    long logSegmentCountBeforeCompaction = state.index.getLogSegmentCount();
    long indexSegmentCountBeforeCompaction = state.index.getIndexSegments().size();
    List<LogSegmentName> allLogSegmentsBeforeCompaction = log.getAllLogSegmentNames();

    Set<Offset> indexSegmentOffsetsUnderCompaction = state.index.getIndexSegments()
        .keySet()
        .stream()
        .filter(offset -> segmentsUnderCompaction.contains(offset.getName()))
        .collect(Collectors.toSet());

    CompactionDetails details = new CompactionDetails(deleteReferenceTimeMs, segmentsUnderCompaction, null);
    long expectedValidDataSize = getValidDataSize(segmentsUnderCompaction, deleteReferenceTimeMs, purgeDeleteTombstone);
    List<LogSegmentName> unaffectedSegments = getUnaffectedSegments(segmentsUnderCompaction);
    // "expiredDeletes" has two parts: 1. expired delete without PUT; 2. expired delete with PUT.
    Pair<Set<MockId>, Set<MockId>> expiredDeletes = new Pair<>(new HashSet<>(), new HashSet<>());
    // get valid log entries including deletes as a backup (in case exception occurred in the middle of compaction)
    List<LogEntry> validLogEntriesInOrder =
        getValidLogEntriesInOrder(segmentsUnderCompaction, deleteReferenceTimeMs, expiredDeletes, purgeDeleteTombstone);
    Set<MockId> compactedDeletes = expiredDeletes.getFirst();
    Set<MockId> deletesWithPuts = expiredDeletes.getSecond();
    Set<MockId> idsInCompactedLogSegments = getIdsWithPutInSegments(segmentsUnderCompaction);
    compactor = getCompactor(log, diskIOScheduler, null, false);
    compactor.initialize(index);

    try {
      compactor.compact(details, bundleReadBuffer);
      if (throwExceptionInsteadOfClose) {
        fail("Compact should have thrown exception");
      }
    } catch (RuntimeException e) {
      assertEquals("Exception not as expected", EXCEPTION_MSG, e.getMessage());
    } finally {
      assertTrue("Close was not induced by the test", closeOrExceptionInduced);
      assertTrue("There should be a compaction in progress",
          CompactionLog.isCompactionInProgress(tempDirStr, STORE_ID));
      compactor.close(0);
    }

    // record any bytes saved because index is going to reinit the metric registry
    long savedBytesReported = metricRegistry.getCounters()
        .get(MetricRegistry.name(BlobStoreCompactor.class, "CompactionBytesReclaimedCount"))
        .getCount();
    // have to reload log since the instance changed by the old compactor is different.
    state.reloadLog(false);
    // use the "real" log, index and disk IO schedulers this time.
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
    // Create a compactor before loading index. Since this compactor would remove the old log segment so the old index segments
    // won't be loaded. This is the same order as the BlobStore initialize the compactor and index.
    state.initIndex(null);
    compactor.initialize(state.index);
    assertEquals("Wrong number of swap segments in use",
        tempDir.list(BlobStoreCompactor.TEMP_LOG_SEGMENTS_FILTER).length, compactor.getSwapSegmentsInUse().length);

    // rebuild compaction history
    CompactionLog.processCompactionLogs(tempDirStr, STORE_ID, STORE_KEY_FACTORY, time, state.config, cLog -> {
      state.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(cLog.getStartTime(),
          cLog.getBeforeAndAfterIndexSegmentOffsetsForCompletedCycles(), false);
      return false;
    });
    state.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();
    // For some reason this compaction is interrupted, we have to check if there is a new log segment added.
    List<LogSegmentName> allLogSegmentsAfterCompaction = log.getAllLogSegmentNames();
    List<LogSegmentName> newLogSegments = allLogSegmentsAfterCompaction.stream()
        .filter(name -> !allLogSegmentsBeforeCompaction.contains(name))
        .collect(Collectors.toList());

    if (newLogSegments.size() != 0) {
      // If we have new log segments, then the before offsets belong to the segments under compaction
      // and the after offsets belong to the new log segments.
      Map<Offset, Offset> indexSegmentOffsets = state.index.getBeforeAndAfterCompactionIndexSegmentOffsets();
      assertTrue(indexSegmentOffsets.size() > 0);
      for (Map.Entry<Offset, Offset> entry : indexSegmentOffsets.entrySet()) {
        assertTrue(segmentsUnderCompaction.contains(entry.getKey().getName()));
        assertTrue(newLogSegments.contains(entry.getValue().getName()));
      }
    }

    try {
      if (CompactionLog.isCompactionInProgress(tempDirStr, STORE_ID)) {
        compactor.resumeCompaction(bundleReadBuffer);
      }
    } finally {
      compactor.close(0);
    }

    verifyIndexSegmentOffsetsBeforeAndAfterCompaction(indexSegmentOffsetsUnderCompaction);
    Set<MockId> remainingBlobIds = getCurrentBlobIdsFromWholeIndex(state.index, null, purgeDeleteTombstone);
    // find those deletes initially had PUTs but were compacted in the subsequent compaction cycle.
    deletesWithPuts.removeAll(remainingBlobIds);

    // remove expired deletes that were compacted in second round of compaction (i.e. resume compaction after shutdown/crash)
    expectedValidDataSize -= deletesWithPuts.size() * DELETE_RECORD_SIZE;
    // some deletes had PUTs before compaction and were considered valid. After first cycle of compaction, the original
    // PUT is compacted (i.e. PUT and DELETE are in different log segments), and the delete may be eligible to be
    // compacted in the second cycle. If this did happen, remove such deletes from valid log entries.
    validLogEntriesInOrder = validLogEntriesInOrder.stream()
        .filter(logEntry -> !deletesWithPuts.contains(logEntry.getId()))
        .collect(Collectors.toList());
    // add deletes (initially with PUTs) that were cleaned up to the "compactedDeletes" set
    compactedDeletes.addAll(deletesWithPuts);

    // if those deletes (which supposed to be compacted) are still present, use valid log entries with deletes
    verifyCompaction(segmentsUnderCompaction, unaffectedSegments, expectedValidDataSize, validLogEntriesInOrder,
        idsInCompactedLogSegments, deleteReferenceTimeMs, compactedDeletes);
    checkVitals(changeExpected, logSegmentSizeSumBeforeCompaction, logSegmentCountBeforeCompaction,
        indexSegmentCountBeforeCompaction);
    if (checkSavedBytesReported) {
      verifySavedBytesCount(logSegmentCountBeforeCompaction, savedBytesReported);
    }
  }
Similar to {@link #compactAndVerify(List, long, boolean)} but expects an interruption and performs recovery before
resuming compaction.
@param log the {@link Log} instance to use for the compaction before interruption.
@param diskIOScheduler the {@link DiskIOScheduler} instance to use for the compaction before interruption.
@param index the {@link PersistentIndex} instance to use for the compaction before interruption.
@param segmentsUnderCompaction the names of the log segments under compaction.
@param deleteReferenceTimeMs the reference time in ms to use to decide whether deletes are valid.
@param changeExpected {@code true} if compaction will cause a change in size of data. {@code false} otherwise.
@param checkSavedBytesReported {@code true} if the metric reporting saved bytes has to be checked. {@code false}
                                           otherwise.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java