private void compactAndVerify(List<LogSegmentName> segmentsUnderCompaction, long deleteReferenceTimeMs,
      boolean changeExpected) throws Exception {
    long logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
    long logSegmentCountBeforeCompaction = state.index.getLogSegmentCount();
    long indexSegmentCountBeforeCompaction = state.index.getIndexSegments().size();

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

    ScheduledExecutorService scheduler = Utils.newScheduler(1, true);
    BlobStoreStats stats =
        new BlobStoreStats("", state.index, 0, Time.MsPerSec, 0, 100, Time.SecsPerMin, false, purgeDeleteTombstone,
            state.time, scheduler, scheduler, DISK_IO_SCHEDULER, new StoreMetrics(new MetricRegistry()), 1, false);
    NavigableMap<LogSegmentName, Long> validDataSizeFromBlobStoreStats =
        stats.getValidDataSizeByLogSegment(new TimeRange(deleteReferenceTimeMs, 0L),
            getFileSpanForLogSegments(segmentsUnderCompaction)).getSecond();
    scheduler.shutdown();
    long totalSizeAfterCompactionFromBlobStoreStats =
        segmentsUnderCompaction.stream().mapToLong(validDataSizeFromBlobStoreStats::get).sum();

    CompactionDetails details = new CompactionDetails(deleteReferenceTimeMs, segmentsUnderCompaction, null);
    long expectedValidDataSize = getValidDataSize(segmentsUnderCompaction, deleteReferenceTimeMs, purgeDeleteTombstone);
    assertEquals("Valid size from blob store should be the same as compacted size", expectedValidDataSize,
        totalSizeAfterCompactionFromBlobStoreStats);
    List<LogSegmentName> unaffectedSegments = getUnaffectedSegments(segmentsUnderCompaction);
    Pair<Set<MockId>, Set<MockId>> expiredDeletes = new Pair<>(new HashSet<>(), new HashSet<>());
    List<LogEntry> validLogEntriesInOrder =
        getValidLogEntriesInOrder(segmentsUnderCompaction, deleteReferenceTimeMs, expiredDeletes, purgeDeleteTombstone);
    Set<MockId> idsInCompactedLogSegments = getIdsWithPutInSegments(segmentsUnderCompaction);
    // "compactedDeletes" are those tombstones that should be compacted in single run (if no exception occurs);
    // "deletesWithPuts" are those tombstones temporarily with PUTs but may be eligible to be compacted in subsequent cycle
    Set<MockId> compactedDeletes = expiredDeletes.getFirst();
    Set<MockId> deletesWithPuts = expiredDeletes.getSecond();

    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
    compactor.initialize(state.index);

    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }
    // First, verify the index segment offset map is good
    verifyIndexSegmentOffsetsBeforeAndAfterCompaction(indexSegmentOffsetsUnderCompaction);

    Set<MockId> remainingBlobIds = getCurrentBlobIdsFromWholeIndex(state.index, null, purgeDeleteTombstone);
    // since this method aims to verify success compaction case, we only need to account for some deletes with PUTs are
    // compacted in the multi-cycle compaction (i.e. PUT is 1st log segment and gets compacted in 1st cycle. DELETE is
    // in 2nd log segment and the 2nd cycle compaction may compact the DELETE as well because source index is updated
    // when switching out the 1st log segment and original PUT is not found in source index)
    deletesWithPuts.removeAll(remainingBlobIds);
    expectedValidDataSize -= deletesWithPuts.size() * DELETE_RECORD_SIZE;
    compactedDeletes.addAll(deletesWithPuts);
    // remove these deletes from valid log entries (if any)
    validLogEntriesInOrder = deletesWithPuts.isEmpty() ? validLogEntriesInOrder : validLogEntriesInOrder.stream()
        .filter(logEntry -> !deletesWithPuts.contains(logEntry.getId()))
        .collect(Collectors.toList());

    assertFalse("No compaction should be in progress", CompactionLog.isCompactionInProgress(tempDirStr, STORE_ID));
    assertEquals("Swap segments should not be found", 0, compactor.getSwapSegmentsInUse().length);
    long logSegmentSizeAfterCompaction = getSumOfLogSegmentEndOffsets();
    long logSegmentCountAfterCompaction = state.index.getLogSegmentCount();
    long indexSegmentCountAfterCompaction = state.index.getIndexSegments().size();
    verifyCompaction(segmentsUnderCompaction, unaffectedSegments, expectedValidDataSize, validLogEntriesInOrder,
        idsInCompactedLogSegments, deleteReferenceTimeMs, compactedDeletes);

    state.reloadLog(true);
    verifyCompaction(segmentsUnderCompaction, unaffectedSegments, expectedValidDataSize, validLogEntriesInOrder,
        idsInCompactedLogSegments, deleteReferenceTimeMs, compactedDeletes);

    assertEquals("Sum of log segment capacities changed after reload", logSegmentSizeAfterCompaction,
        getSumOfLogSegmentEndOffsets());
    assertEquals("Log segment count changed after reload", logSegmentCountAfterCompaction,
        state.index.getLogSegmentCount());
    assertEquals("Index segment count changed after reload", indexSegmentCountAfterCompaction,
        state.index.getIndexSegments().size());
    checkVitals(changeExpected, logSegmentSizeSumBeforeCompaction, logSegmentCountBeforeCompaction,
        indexSegmentCountBeforeCompaction);
    verifySavedBytesCount(logSegmentCountBeforeCompaction, 0);
  }
Compacts the {@code segmentsUnderCompaction} and verifies sanity of store and data. Also verifies that no change
occurred if no change was expected and vice versa. Ensures that reloading the log does not have effects on the
integrity of the store and data.
@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.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java