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

    // Log Segment 0
    // Index Segment 0.1 P D D -> P D
    IndexEntry p1 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.addDeleteEntry((MockId) p1.getKey());
    state.addDeleteEntry((MockId) p1.getKey(), null, (short) 1);
    state.addPutEntries(2, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.2 P T D U D -> P T D
    IndexEntry p2 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.makePermanent((MockId) p2.getKey(), false);
    state.addDeleteEntry((MockId) p2.getKey());
    state.addUndeleteEntry((MockId) p2.getKey());
    state.addDeleteEntry((MockId) p2.getKey());

    // Index Segment 0.3 P U U D -> P D
    IndexEntry p3 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.addUndeleteEntry((MockId) p3.getKey(), (short) 1);
    state.addUndeleteEntry((MockId) p3.getKey(), (short) 2);
    state.addDeleteEntry((MockId) p3.getKey());
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    writeDataToMeetRequiredSegmentCount(2, null);
    state.reloadIndex(true, false);

    LogSegmentName logSegmentName = p1.getValue().getOffset().getName();
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 1);
    long endOffsetOfSegmentBeforeCompaction = state.log.getSegment(logSegmentName).getEndOffset();
    // Use 0 as delete reference time so all the deletes are still with in the retention date
    CompactionDetails details = new CompactionDetails(0, segmentsUnderCompaction, null);
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
    compactor.initialize(state.index);

    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }
    // Before compaction, the records in the log are
    // P1 D1 D1 P P| P2 T2 D2 U2 D| P3 U3 U3 D3 P
    // After compaction, the records in the log are
    // P1 D1 P P P2| T2 D2 P3 D3 P

    long ds = DELETE_RECORD_SIZE;
    long ps = PUT_RECORD_SIZE;
    long us = UNDELETE_RECORD_SIZE;
    long ts = TTL_UPDATE_RECORD_SIZE;
    long cleanedUpSize = ds + (ds + us) + 2 * us;
    LogSegmentName compactedLogSegmentName = logSegmentName.getNextGenerationName();
    assertEquals("End offset of log segment not as expected after compaction",
        endOffsetOfSegmentBeforeCompaction - cleanedUpSize,
        state.log.getSegment(compactedLogSegmentName).getEndOffset());

    LogSegment compactedLogSegment = state.log.getSegment(compactedLogSegmentName);
    ConcurrentNavigableMap<Offset, IndexSegment> indexSegments = state.index.getIndexSegments();
    FindEntriesCondition condition = new FindEntriesCondition(Long.MAX_VALUE);
    long currentExpectedOffset = compactedLogSegment.getStartOffset();

    // Get the entries in the first segment
    IndexSegment segment = indexSegments.get(new Offset(compactedLogSegmentName, compactedLogSegment.getStartOffset()));
    List<IndexEntry> indexEntries = new ArrayList<>();
    assertEquals("LogSegment name mismatch", compactedLogSegmentName, segment.getStartOffset().getName());
    segment.getIndexEntriesSince(null, condition, indexEntries, new AtomicLong(0), false, false);
    Collections.sort(indexEntries, PersistentIndex.INDEX_ENTRIES_OFFSET_COMPARATOR);
    verifyIndexEntry(indexEntries.get(0), (MockId) p1.getKey(), currentExpectedOffset, ps, Utils.Infinite_Time, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(1), (MockId) p1.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        false, false, (short) 1);
    currentExpectedOffset += ds + 2 * ps; // skip two puts
    verifyIndexEntry(indexEntries.get(4), (MockId) p2.getKey(), currentExpectedOffset, ps, Utils.Infinite_Time, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;

    // Get the entries in the second segment
    segment = indexSegments.higherEntry(segment.getStartOffset()).getValue();
    indexEntries.clear();
    assertEquals("LogSegment name mismatch", compactedLogSegmentName, segment.getStartOffset().getName());
    segment.getIndexEntriesSince(null, condition, indexEntries, new AtomicLong(0), false, false);
    Collections.sort(indexEntries, PersistentIndex.INDEX_ENTRIES_OFFSET_COMPARATOR);
    verifyIndexEntry(indexEntries.get(0), (MockId) p2.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 0);
    currentExpectedOffset += ts;
    verifyIndexEntry(indexEntries.get(1), (MockId) p2.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        true, false, (short) 1);
    currentExpectedOffset += ds;
    verifyIndexEntry(indexEntries.get(2), (MockId) p3.getKey(), currentExpectedOffset, ps, Utils.Infinite_Time, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(3), (MockId) p3.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        false, false, (short) 2);

    // 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);
  }
Test when the PUT, DELETE, TTL_UPDATE and UNDELETE of the same key are in the same index segment and the final state
is delete but it's still within the retention date.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java