@Test
  public void compactWholeLogMultipleTimesTest() throws Exception {
    refreshState(false, true, false);
    long requiredCount = state.log.getCapacityInBytes() / state.log.getSegmentCapacity() - 3;
    long expiryTimeMs = getInvalidationTime(requiredCount);
    List<Long> expiryTimesMs = Arrays.asList(state.time.milliseconds() / 2, expiryTimeMs, expiryTimeMs * 2);
    writeDataToMeetRequiredSegmentCount(requiredCount, expiryTimesMs);
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    Set<MockId> compactedDeletes = new HashSet<>();
    Set<MockId> idsInCompactedLogSegments = getIdsWithPutInSegments(segmentsUnderCompaction);

    for (long setTimeMs : expiryTimesMs) {
      if (state.time.milliseconds() < setTimeMs + Time.MsPerSec) {
        state.advanceTime(setTimeMs + Time.MsPerSec - state.time.milliseconds());
      }
      long deleteReferenceTimeMs = state.time.milliseconds();
      getCurrentBlobIdsFromWholeIndex(state.index, compactedDeletes, purgeDeleteTombstone);
      long logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
      CompactionDetails details = new CompactionDetails(deleteReferenceTimeMs, segmentsUnderCompaction, null);

      compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
      compactor.initialize(state.index);
      long logSegmentsBeforeCompaction = state.index.getLogSegmentCount();
      try {
        compactor.compact(details, bundleReadBuffer);
      } finally {
        compactor.close(0);
      }
      assertFalse("Sum of size of log segments did not change after compaction",
          logSegmentSizeSumBeforeCompaction == getSumOfLogSegmentEndOffsets());
      verifyDataPostCompaction(idsInCompactedLogSegments, compactedDeletes, deleteReferenceTimeMs);
      state.reloadLog(true);
      verifyDataPostCompaction(idsInCompactedLogSegments, compactedDeletes, deleteReferenceTimeMs);
      segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
      state.verifyRealIndexSanity();
      // no clean shutdown file should exist
      assertFalse("Clean shutdown file not deleted",
          new File(tempDirStr, BlobStoreCompactor.TARGET_INDEX_CLEAN_SHUTDOWN_FILE_NAME).exists());
      // there should be no temp files
      assertEquals("There are some temp log segments", 0,
          tempDir.listFiles(BlobStoreCompactor.TEMP_LOG_SEGMENTS_FILTER).length);
      verifySavedBytesCount(logSegmentsBeforeCompaction, 0);
    }
  }
Compacts the whole log multiple times with some data compacted each time.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java