@Test
  public void undeleteSameIndexSegmentTest_NoPut() 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 P P P P
    state.addPutEntries(5, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.2 D -> D
    MockId p2 = state.getUniqueId();
    state.addDeleteEntry(p2, new MessageInfo(p2, DELETE_RECORD_SIZE, true, false, p2.getAccountId(), p2.getAccountId(),
        state.time.milliseconds()), (short) 0);
    state.addPutEntries(4, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.3 U -> U
    MockId p3 = state.getUniqueId();
    state.addUndeleteEntry(p3,
        new MessageInfo(p3, UNDELETE_RECORD_SIZE, true, false, p3.getAccountId(), p3.getAccountId(),
            state.time.milliseconds()), (short) 1);
    state.addPutEntries(4, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.4 T, U -> T, U
    MockId p4 = state.getUniqueId();
    state.makePermanent(p4, true);
    state.addUndeleteEntry(p4,
        new MessageInfo(p4, UNDELETE_RECORD_SIZE, true, false, p4.getAccountId(), p4.getAccountId(),
            state.time.milliseconds()), (short) 1);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.5 T, D -> D
    MockId p5 = state.getUniqueId();
    state.makePermanent(p5, true);
    state.addDeleteEntry(p5, null, (short) 0);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.6 D, U -> U
    MockId p6 = state.getUniqueId();
    state.addDeleteEntry(p6, new MessageInfo(p6, DELETE_RECORD_SIZE, true, false, p6.getAccountId(), p6.getAccountId(),
        state.time.milliseconds()), (short) 0);
    state.addUndeleteEntry(p6,
        new MessageInfo(p6, UNDELETE_RECORD_SIZE, true, false, p6.getAccountId(), p6.getAccountId(),
            state.time.milliseconds()), (short) 1);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.7 U, D -> D
    MockId p7 = state.getUniqueId();
    state.addUndeleteEntry(p7,
        new MessageInfo(p7, UNDELETE_RECORD_SIZE, true, false, p7.getAccountId(), p7.getAccountId(),
            state.time.milliseconds()), (short) 1);
    state.addDeleteEntry(p7, null, (short) 1);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.8 T, D, U -> T, U
    MockId p8 = state.getUniqueId();
    state.makePermanent(p8, true);
    state.addDeleteEntry(p8, null, (short) 0);
    state.addUndeleteEntry(p8,
        new MessageInfo(p8, UNDELETE_RECORD_SIZE, true, false, p8.getAccountId(), p8.getAccountId(),
            state.time.milliseconds()), (short) 1);
    state.addPutEntries(2, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.9 D, U, T -> U, T
    MockId p9 = state.getUniqueId();
    state.addDeleteEntry(p9, new MessageInfo(p9, DELETE_RECORD_SIZE, true, false, p9.getAccountId(), p9.getAccountId(),
        state.time.milliseconds()), (short) 0);
    state.addUndeleteEntry(p9,
        new MessageInfo(p9, UNDELETE_RECORD_SIZE, true, false, p9.getAccountId(), p9.getAccountId(),
            state.time.milliseconds()), (short) 1);
    state.makePermanent(p9, false);
    state.addPutEntries(2, PUT_RECORD_SIZE, Utils.Infinite_Time);

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

    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 1);
    LogSegmentName logSegmentName = segmentsUnderCompaction.get(0);
    long endOffsetOfSegmentBeforeCompaction = state.log.getSegment(logSegmentName).getEndOffset();
    CompactionDetails details = new CompactionDetails(state.time.milliseconds(), 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
    // P P P P P| D2 P P P P| U3 P P P P| T4 U4 P P P| T5 D5 P P P| D6 U6 P P P| U7 D7 P P P| T8 D8 U8 P P| D9 U9 T9 P P
    // After compaction, the records in the log are
    // P P P P P| D2 P P P P| U3 P P P P| T4 U4 P P P| D5 P P P U6| P P P D7 P| P P T8 U8 P| P U9 T9 P P

    long ds = DELETE_RECORD_SIZE;
    long ps = PUT_RECORD_SIZE;
    long us = UNDELETE_RECORD_SIZE;
    long ts = TTL_UPDATE_RECORD_SIZE;
    long cleanedUpSize = 0 + 0 + 0 + 0 + ts + ds + us + ds + ds;
    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<>();
    currentExpectedOffset += 5 * ps; // skip five puts

    // 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), p2, currentExpectedOffset, ds, Utils.Infinite_Time, true, false, false,
        (short) 0);
    currentExpectedOffset += ds + 4 * ps; // skip four puts

    // Get the entries in the third 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), p3, currentExpectedOffset, us, Utils.Infinite_Time, false, false, true,
        (short) 1);
    currentExpectedOffset += us + 4 * ps; // skip four puts

    // Get the entries in the fourth 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), p4, currentExpectedOffset, ts, Utils.Infinite_Time, false, true, false,
        (short) 0);
    currentExpectedOffset += ts;
    verifyIndexEntry(indexEntries.get(1), p4, currentExpectedOffset, us, Utils.Infinite_Time, false, true, true,
        (short) 1);
    currentExpectedOffset += us + 3 * ps; // skip three puts

    // Get the entries in the fifth 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), p5, currentExpectedOffset, ds, Utils.Infinite_Time, true, true, false,
        (short) 0);
    currentExpectedOffset += ds + 3 * ps; // skip three puts
    verifyIndexEntry(indexEntries.get(4), p6, currentExpectedOffset, us, Utils.Infinite_Time, false, false, true,
        (short) 1);
    currentExpectedOffset += us;

    // Get the entries in the sixth 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);
    currentExpectedOffset += 3 * ps; // skip three puts
    verifyIndexEntry(indexEntries.get(3), p7, currentExpectedOffset, ds, Utils.Infinite_Time, true, false, false,
        (short) 1);
    currentExpectedOffset += ds + ps; // skip one put

    // Get the entries in the seventh 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);
    currentExpectedOffset += 2 * ps; // skip two puts
    verifyIndexEntry(indexEntries.get(2), p8, currentExpectedOffset, ts, Utils.Infinite_Time, false, true, false,
        (short) 0);
    currentExpectedOffset += ts;
    verifyIndexEntry(indexEntries.get(3), p8, currentExpectedOffset, us, Utils.Infinite_Time, false, true, true,
        (short) 1);
    currentExpectedOffset += us + ps; // skip one put

    // Get the entries in the eighth 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);
    currentExpectedOffset += ps; // skip one put
    verifyIndexEntry(indexEntries.get(1), p9, currentExpectedOffset, us, Utils.Infinite_Time, false, false, true,
        (short) 1);
    currentExpectedOffset += us;
    verifyIndexEntry(indexEntries.get(2), p9, currentExpectedOffset, ts, Utils.Infinite_Time, false, true, false,
        (short) 1);

    // 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 there is no PUT in the beginning of the blob history and all the records for the same key are in the same
index segment. It's possible that PUT record is compacted in a previous compaction cycle.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java