@Test
  public void containerDeletionTest() throws Exception {
    refreshState(false, false, false);

    state.properties.setProperty("store.container.deletion.enabled", "true");
    state.properties.setProperty("store.index.max.number.of.inmem.elements", "5");
    state.initIndex(null);
    long notExpiredMs = state.time.milliseconds() + TimeUnit.SECONDS.toMillis(Short.MAX_VALUE);
    // LS (Log Segment) 0
    // IS (Index Segment) 0.1
    // p1 DeleteInProgress
    // p2 Inactive
    IndexEntry p1 = state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);
    IndexEntry p2 = state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);
    state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);
    state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);
    state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);

    // IS (Index Segment) 0.2
    // p3 DeleteInProgress
    // p4 Inactive
    IndexEntry p3 = state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);
    IndexEntry p4 = state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);
    long lastRecSize = state.log.getSegmentCapacity() - state.index.getCurrentEndOffset().getOffset();
    state.addPutEntries(1, lastRecSize, notExpiredMs).get(0);

    // LS 1
    state.addPutEntries(1, PUT_RECORD_SIZE, notExpiredMs).get(0);

    // Make sure we have two log segments
    writeDataToMeetRequiredSegmentCount(2, null);

    //add to be deleted index entries into indexEntriesDeleteInProgress list.
    List<IndexEntry> indexEntriesDeleteInProgress = new ArrayList<>();
    indexEntriesDeleteInProgress.add(p1);
    indexEntriesDeleteInProgress.add(p3);
    //add invalid index entries into indexEntriesInvalid list.
    List<IndexEntry> indexEntriesInactive = new ArrayList<>();
    indexEntriesInactive.add(p2);
    indexEntriesInactive.add(p4);

    // Mock indexEntries' corresponding container as INACTIVE/DELETE_IN_PROGRESS.
    Set<Container> deleteInProgressSet = new HashSet<>();
    Set<MockId> deletedInProgressKeys = new HashSet<>();
    Set<Container> InactiveSet = new HashSet<>();
    Set<MockId> InactiveKeys = new HashSet<>();
    long cleanedUpSize = 0;

    for (IndexEntry indexEntry : indexEntriesDeleteInProgress) {
      Container container = Mockito.mock(Container.class);
      Mockito.when(container.getParentAccountId()).thenReturn(indexEntry.getValue().getAccountId());
      Mockito.when(container.getId()).thenReturn(indexEntry.getValue().getContainerId());
      Mockito.when(container.getDeleteTriggerTime())
          .thenReturn(System.currentTimeMillis() - TimeUnit.DAYS.toMillis(config.storeContainerDeletionRetentionDays));
      deleteInProgressSet.add(container);
      deletedInProgressKeys.add((MockId) indexEntry.getKey());
      cleanedUpSize += indexEntry.getValue().getSize();
    }

    for (IndexEntry indexEntry : indexEntriesInactive) {
      Container container = Mockito.mock(Container.class);
      Mockito.when(container.getParentAccountId()).thenReturn(indexEntry.getValue().getAccountId());
      Mockito.when(container.getId()).thenReturn(indexEntry.getValue().getContainerId());
      InactiveSet.add(container);
      InactiveKeys.add((MockId) indexEntry.getKey());
      cleanedUpSize += indexEntry.getValue().getSize();
    }

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

    // get everything except the last log segment entries out of the journal
    state.reloadIndex(true, false);
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 1);
    long logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
    CompactionDetails details = new CompactionDetails(notExpiredMs, segmentsUnderCompaction, null);

    LogSegmentName logSegmentName = p1.getValue().getOffset().getName();
    LogSegmentName compactedLogSegmentName = logSegmentName.getNextGenerationName();
    long endOffsetOfSegmentBeforeCompaction = state.log.getSegment(logSegmentName).getEndOffset();

    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
    compactor.initialize(state.index);
    int indexSegmentCountBeforeCompaction = state.index.getIndexSegments().size();

    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }

    assertFalse("Sum of size of log segments did not change after compaction",
        logSegmentSizeSumBeforeCompaction == getSumOfLogSegmentEndOffsets());

    // inactive/deleted indexEntries should not be found
    for (MockId deletedKey : deletedInProgressKeys) {
      assertNull("There should be no record of " + deletedKey, state.index.findKey(deletedKey));
    }

    for (MockId inactiveKey : InactiveKeys) {
      assertNull("There should be no record of " + inactiveKey, state.index.findKey(inactiveKey));
    }

    assertEquals("End offset of log segment not as expected after compaction",
        endOffsetOfSegmentBeforeCompaction - cleanedUpSize,
        state.log.getSegment(compactedLogSegmentName).getEndOffset());

    int indexSegmentDiff = 1;
    assertEquals("Index Segment not as expected after compaction", indexSegmentCountBeforeCompaction - indexSegmentDiff,
        state.index.getIndexSegments().size());
  }
Tests compaction on a log that contains a PUT record which the corresponding container marked as INACTIVE/DELETE_IN_PROGRESS
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java