@Test
  public void testCloseLastLogSegmentIfQualified() throws Exception {
    //create first active log segment
    refreshState(false, false, true);
    //leave five log segment space for auto close last log segment purpose
    long requiredCount = state.log.getCapacityInBytes() / state.log.getSegmentCapacity() - 5;
    writeDataToMeetRequiredSegmentCountForAutoCloseLogSegmentTest(requiredCount);

    //delete blobs in last index segment.
    ConcurrentSkipListMap<Offset, IndexSegment> AllIndexSegments = state.index.getIndexSegments();
    IndexSegment indexSegment = AllIndexSegments.floorEntry(state.log.getEndOffset()).getValue();
    ListIterator<IndexEntry> segmentListIter = indexSegment.listIterator(indexSegment.size());

    Set<Container> deleteInProgressSet = new HashSet<>();
    int cnt = 0;
    while (segmentListIter.hasNext()) {
      IndexEntry indexEntry = segmentListIter.next();
      MockId mockId = (MockId) indexEntry.getKey();
      Container container = new ContainerBuilder(mockId.getContainerId(), "containerName" + cnt++,
          Container.ContainerStatus.DELETE_IN_PROGRESS, null, mockId.getAccountId()).build();
      deleteInProgressSet.add(container);
    }

    Mockito.when(accountService.getContainersByStatus(Container.ContainerStatus.DELETE_IN_PROGRESS))
        .thenReturn(deleteInProgressSet);

    // reload index to make sure journal is on only the latest log segment
    state.reloadIndex(true, false);
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    // Prepare the compactionPolicyInfo before compaction.
    CompactionPolicySwitchInfo compactionPolicySwitchInfo =
        new CompactionPolicySwitchInfo(System.currentTimeMillis(), true);
    backUpCompactionPolicyInfo(tempDir.toString(), compactionPolicySwitchInfo);
    //instantiate compactor.
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, true);
    compactor.initialize(state.index);

    compactAndVerifyForContainerDeletion(segmentsUnderCompaction, state.time.milliseconds(), false);
    //increment the counter
    compactionPolicySwitchInfo.setNextRoundIsCompactAllPolicy(false);
    // reload index to make sure journal is on only the latest log segment
    backUpCompactionPolicyInfo(tempDir.toString(), compactionPolicySwitchInfo);
    segmentsUnderCompaction = getLogSegments(0,
        state.log.getCapacityInBytes() / state.log.getSegmentCapacity() - state.log.getRemainingUnallocatedSegments()
            - 1);

    compactAndVerifyForContainerDeletion(segmentsUnderCompaction, state.time.milliseconds(), true);

    // Edge case test: if the last log segment is empty, no need to auto close it.
    int beforeAutoCloseLogSegmentsCnt = state.log.getLogSegmentCount();
    compactionPolicySwitchInfo = new CompactionPolicySwitchInfo(System.currentTimeMillis(), true);
    backUpCompactionPolicyInfo(tempDir.toString(), compactionPolicySwitchInfo);
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, true);
    compactor.initialize(state.index);
    if (state.log.autoCloseLastLogSegmentIfQualified()) {
      //refresh journal.
      state.index.journal.cleanUpJournal();
    }
    int afterAutoCloseLogSegmentsCnt = state.log.getLogSegmentCount();
    assertEquals("No segments should be created since last log segment is empty", beforeAutoCloseLogSegmentsCnt,
        afterAutoCloseLogSegmentsCnt);

    //make sure new data will be added into the last log segment
    long logSegmentCountBeforeNewDataAddIntoAutoClosedLogSegment = state.index.getLogSegmentCount();
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);
    long logSegmentCountAfterNewDataAddIntoAutoClosedLogSegment = state.index.getLogSegmentCount();
    assertEquals("Log Segment count should be increased after some data has been added to the last log segment.",
        logSegmentCountBeforeNewDataAddIntoAutoClosedLogSegment,
        logSegmentCountAfterNewDataAddIntoAutoClosedLogSegment - 1);
    assertEquals("Last index segment should belongs to auto closed log segment",
        state.index.getIndexSegments().lastEntry().getValue().getLogSegmentName(),
        state.log.getLastSegment().getName());

    //close the last log segment again.
    beforeAutoCloseLogSegmentsCnt = state.log.getLogSegmentCount();
    if (state.log.autoCloseLastLogSegmentIfQualified()) {
      //refresh journal.
      state.index.journal.cleanUpJournal();
    }
    afterAutoCloseLogSegmentsCnt = state.log.getLogSegmentCount();
    assertEquals("One log segment should be created since last log segment is not empty", beforeAutoCloseLogSegmentsCnt,
        afterAutoCloseLogSegmentsCnt - 1);

    //make sure new data will be added into the last log segment
    logSegmentCountBeforeNewDataAddIntoAutoClosedLogSegment = state.index.getLogSegmentCount();
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);
    logSegmentCountAfterNewDataAddIntoAutoClosedLogSegment = state.index.getLogSegmentCount();
    assertEquals("Log Segment count should be increased after some data has been added to the last log segment.",
        logSegmentCountBeforeNewDataAddIntoAutoClosedLogSegment,
        logSegmentCountAfterNewDataAddIntoAutoClosedLogSegment - 1);
    assertEquals("Last index segment should belongs to auto closed log segment",
        state.index.getIndexSegments().lastEntry().getValue().getLogSegmentName(),
        state.log.getLastSegment().getName());

    // check flow after deployment
    state.reloadLog(true);
  }
Does compaction tests where compaction is interrupted (and recovered) after copying a few records from an index
segment.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java