@Test
  public void interspersedDeletedAndExpiredBlobsTest() throws Exception {
    refreshState(false, false, false);
    state.properties.put("store.index.max.number.of.inmem.elements", Integer.toString(5));
    state.initIndex(null);

    int numFinalSegmentsCount = 3;
    long expiryTimeMs = getInvalidationTime(numFinalSegmentsCount + 1);
    // fill up one segment.
    writeDataToMeetRequiredSegmentCount(1, null);

    // IS 1.1 starts
    // 1. Put entry that contains a delete entry in the same index segment and is not counted as deleted.
    // won't be cleaned up.
    IndexEntry entry = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    assertEquals("There should have been a new log segment created", 2, state.index.getLogSegmentCount());
    LogSegmentName logSegmentName = entry.getValue().getOffset().getName();
    MockId delUnexpPutSameIdxSegId = (MockId) entry.getKey();
    state.addDeleteEntry(delUnexpPutSameIdxSegId);

    // 2. Put entry that has expired and contains a delete entry in the same index segment. Does not count as deleted
    // but is expired.
    // will be cleaned up, but delete record remains
    MockId delExpPutSameIdxSegId = (MockId) state.addPutEntries(1, PUT_RECORD_SIZE, 0).get(0).getKey();
    state.addDeleteEntry(delExpPutSameIdxSegId);

    // 3. Put entry that will be deleted.
    // won't be cleaned up.
    MockId willBeDelPut = (MockId) state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0).getKey();
    // IS 1.1 ends

    // roll over. IS 1.2 starts
    // 4. Put entry that has expired.
    // will be cleaned up.
    state.addPutEntries(1, PUT_RECORD_SIZE, 0).get(0).getKey();

    // 5. Put entry with an expiry time  that is not expired and has a delete entry in the same index segment and is not
    // counted as deleted or expired. In the compacted log, the put entry will be in one index segment and the delete
    // in another
    // won't be cleaned up.
    MockId delUnexpPutDiffIdxSegId = (MockId) state.addPutEntries(1, PUT_RECORD_SIZE, expiryTimeMs).get(0).getKey();
    state.addDeleteEntry(delUnexpPutDiffIdxSegId);

    // 6. Put entry that will expire (but is valid right now).
    // won't be cleaned up.
    MockId willExpPut = (MockId) state.addPutEntries(1, PUT_RECORD_SIZE, expiryTimeMs).get(0).getKey();

    // 7. Delete entry for an id that is in another index segment
    // won't be cleaned up.
    state.addDeleteEntry(willBeDelPut);
    // IS 1.2 ends

    // rollover. IS 1.3 starts
    // 8. Delete entry for an id that is in another log segment
    // won't be cleaned up.
    MockId idFromAnotherSegment = state.getIdToDeleteFromLogSegment(state.log.getFirstSegment(), false);
    state.addDeleteEntry(idFromAnotherSegment);

    // 9. Delete entry for a Put entry that doesn't exist. However, if it existed, it wouldn't have been eligible for
    // cleanup
    // the delete record itself won't be cleaned up
    MockId uniqueId = state.getUniqueId();
    state.addDeleteEntry(uniqueId,
        new MessageInfo(uniqueId, Integer.MAX_VALUE, expiryTimeMs, uniqueId.getAccountId(), uniqueId.getContainerId(),
            state.time.milliseconds()));

    // fill up the rest of the segment + one more
    writeDataToMeetRequiredSegmentCount(numFinalSegmentsCount, null);
    // reload index to keep the journal only in the last log segment
    state.reloadIndex(true, false);

    long deleteReferenceTimeMs = 0;
    List<LogSegmentName> segmentsUnderCompaction = Collections.singletonList(logSegmentName);
    long endOffsetOfSegmentBeforeCompaction = state.log.getSegment(logSegmentName).getEndOffset();
    CompactionDetails details = new CompactionDetails(deleteReferenceTimeMs, segmentsUnderCompaction, null);
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
    compactor.initialize(state.index);
    long logSegmentCountBeforeCompaction = state.index.getLogSegmentCount();
    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }

    LogSegmentName compactedLogSegmentName = logSegmentName.getNextGenerationName();
    LogSegment compactedLogSegment = state.log.getSegment(compactedLogSegmentName);
    long cleanedUpSize = 2 * PUT_RECORD_SIZE;
    assertEquals("End offset of log segment not as expected after compaction",
        endOffsetOfSegmentBeforeCompaction - cleanedUpSize,
        state.log.getSegment(compactedLogSegmentName).getEndOffset());

    FindEntriesCondition condition = new FindEntriesCondition(Long.MAX_VALUE);
    // get the first index segment that refers to the compacted segment
    IndexSegment indexSegment =
        state.index.getIndexSegments().get(new Offset(compactedLogSegmentName, compactedLogSegment.getStartOffset()));
    List<IndexEntry> indexEntries = new ArrayList<>();
    assertTrue("Should have got some index entries",
        indexSegment.getIndexEntriesSince(null, condition, indexEntries, new AtomicLong(0), true, false));
    assertEquals("There should be 5 index entries returned", 4, indexEntries.size());
    indexEntries.sort(PersistentIndex.INDEX_ENTRIES_OFFSET_COMPARATOR);

    long logSegmentStartOffset = compactedLogSegment.getStartOffset();
    long currentExpectedOffset = logSegmentStartOffset + PUT_RECORD_SIZE;
    // first index entry should be a delete and it should have an original message offset
    verifyIndexEntry(indexEntries.get(0), delUnexpPutSameIdxSegId, currentExpectedOffset, DELETE_RECORD_SIZE,
        Utils.Infinite_Time, true, logSegmentStartOffset);
    currentExpectedOffset += DELETE_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(1), delExpPutSameIdxSegId, currentExpectedOffset, DELETE_RECORD_SIZE, 0, true,
        IndexValue.UNKNOWN_ORIGINAL_MESSAGE_OFFSET);
    currentExpectedOffset += DELETE_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(2), willBeDelPut, currentExpectedOffset, PUT_RECORD_SIZE, Utils.Infinite_Time,
        false, currentExpectedOffset);
    long willBeDelOffset = currentExpectedOffset;
    currentExpectedOffset += PUT_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(3), delUnexpPutDiffIdxSegId, currentExpectedOffset, PUT_RECORD_SIZE, expiryTimeMs,
        false, currentExpectedOffset);
    currentExpectedOffset += PUT_RECORD_SIZE;

    // get the second index segment
    indexSegment = state.index.getIndexSegments().higherEntry(indexSegment.getStartOffset()).getValue();
    indexEntries.clear();
    assertTrue("Should have got some index entries",
        indexSegment.getIndexEntriesSince(null, condition, indexEntries, new AtomicLong(0), true, false));
    assertEquals("There should be 5 index entries returned", 5, indexEntries.size());
    indexEntries.sort(PersistentIndex.INDEX_ENTRIES_OFFSET_COMPARATOR);
    verifyIndexEntry(indexEntries.get(0), delUnexpPutDiffIdxSegId, currentExpectedOffset, DELETE_RECORD_SIZE,
        expiryTimeMs, true, currentExpectedOffset - PUT_RECORD_SIZE);
    currentExpectedOffset += DELETE_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(1), willExpPut, currentExpectedOffset, PUT_RECORD_SIZE, expiryTimeMs, false,
        currentExpectedOffset);
    currentExpectedOffset += PUT_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(2), willBeDelPut, currentExpectedOffset, DELETE_RECORD_SIZE, Utils.Infinite_Time,
        true, willBeDelOffset);
    currentExpectedOffset += DELETE_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(3), idFromAnotherSegment, currentExpectedOffset, DELETE_RECORD_SIZE,
        Utils.Infinite_Time, true, IndexValue.UNKNOWN_ORIGINAL_MESSAGE_OFFSET);
    currentExpectedOffset += DELETE_RECORD_SIZE;
    verifyIndexEntry(indexEntries.get(4), uniqueId, currentExpectedOffset, DELETE_RECORD_SIZE, expiryTimeMs, true,
        IndexValue.UNKNOWN_ORIGINAL_MESSAGE_OFFSET);

    // 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(logSegmentCountBeforeCompaction, 0);
  }
Tests the case where deletes and expired blobs are interspersed and the expired blobs are eligible for cleanup
but deleted blobs (also includes blobs that have been put and deleted in the same index segment) are not.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java