@Test
  public void undeleteSameIndexSegmentTest_WithTtlUpdate() 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
    long expirationTime = state.time.milliseconds() + 500 * Time.MsPerSec;
    // Index Segment 0.1 P T D U -> P T U
    IndexEntry p1 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.makePermanent((MockId) p1.getKey(), false);
    state.addDeleteEntry((MockId) p1.getKey());
    state.addUndeleteEntry((MockId) p1.getKey());
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

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

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

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

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

    // Index Segment 0.6 P(expired) D U T D -> D
    IndexEntry p6 = state.addPutEntries(1, PUT_RECORD_SIZE, expirationTime).get(0);
    state.addDeleteEntry((MockId) p6.getKey());
    state.addUndeleteEntry((MockId) p6.getKey());
    state.makePermanent((MockId) p6.getKey(), false);
    state.addDeleteEntry((MockId) p6.getKey());

    // Index Segment 0.7 P T -> P T
    IndexEntry p7 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.makePermanent((MockId) p7.getKey(), false);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.8 P(expired) T -> P T
    IndexEntry p8 = state.addPutEntries(1, PUT_RECORD_SIZE, expirationTime).get(0);
    state.makePermanent((MockId) p8.getKey(), false);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 0.9 P U U T -> P U T
    IndexEntry p9 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.addUndeleteEntry((MockId) p9.getKey(), (short) 1);
    state.addUndeleteEntry((MockId) p9.getKey(), (short) 2);
    state.makePermanent((MockId) p9.getKey(), false);
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    writeDataToMeetRequiredSegmentCount(1, null);
    // This is the end of the first log segment

    // Log segment 1
    // Index Segment 1.1 P(expired) U U T -> P U T
    IndexEntry p11 = state.addPutEntries(1, PUT_RECORD_SIZE, expirationTime).get(0);
    state.addUndeleteEntry((MockId) p11.getKey(), (short) 1);
    state.addUndeleteEntry((MockId) p11.getKey(), (short) 2);
    state.makePermanent((MockId) p11.getKey(), false);
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 1.2 P T U U -> P T U
    IndexEntry p12 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.makePermanent((MockId) p12.getKey(), false);
    state.addUndeleteEntry((MockId) p12.getKey(), (short) 1);
    state.addUndeleteEntry((MockId) p12.getKey(), (short) 2);
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 1.3 P(expired) T U U -> P T U
    IndexEntry p13 = state.addPutEntries(1, PUT_RECORD_SIZE, expirationTime).get(0);
    state.makePermanent((MockId) p13.getKey(), false);
    state.addUndeleteEntry((MockId) p13.getKey(), (short) 1);
    state.addUndeleteEntry((MockId) p13.getKey(), (short) 2);
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 1.4 P T D D -> D
    IndexEntry p14 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.makePermanent((MockId) p14.getKey(), false);
    state.addDeleteEntry((MockId) p14.getKey());
    state.addDeleteEntry((MockId) p14.getKey(), null, (short) 1);
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 1.5 P(expired) T D D -> D
    IndexEntry p15 = state.addPutEntries(1, PUT_RECORD_SIZE, expirationTime).get(0);
    state.makePermanent((MockId) p15.getKey(), false);
    state.addDeleteEntry((MockId) p15.getKey());
    state.addDeleteEntry((MockId) p15.getKey(), null, (short) 1);
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Index Segment 1.6 T -> [], Put already compacted
    MockId p16id = state.getUniqueId();
    state.makePermanent(p16id, true);
    state.addPutEntries(4, PUT_RECORD_SIZE, Utils.Infinite_Time);

    // Make sure we have 3 log segments
    writeDataToMeetRequiredSegmentCount(3, null);
    state.reloadIndex(true, false);

    LogSegmentName logSegmentName = p1.getValue().getOffset().getName();
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 1);
    state.advanceTime(expirationTime - state.time.milliseconds() + 1000);
    long deleteReferenceTimeMs = state.time.milliseconds();
    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);

    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }
    // Before compaction, the records in the log are
    // P1 T1 D1 U1 P| P2 T2 D2 U2 P| P3 D3 U3 T3 P| P4 D4 U4 T4 P| P5 D5 U5 T5 D5| P6 D6 U6 T6 D6| P7 T7 P P P| P8 T8 P P P| P9 U9 U9 T9 P
    // After compaction, the records in the log are
    // P1 T1 U1 P P2| T2 U2 P P3 U3| T3 P P4 U4 T4| P D5 D6 P7 T7| P P P P8 T8| P P P P9 U9| T9

    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 + ds + ds + (ps + ds + us + ts) + (ps + ds + us + ts) + 0 + 0 + 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, ts, Utils.Infinite_Time, false,
        true, false, (short) 0);
    currentExpectedOffset += ts;
    verifyIndexEntry(indexEntries.get(2), (MockId) p1.getKey(), currentExpectedOffset, us, Utils.Infinite_Time, false,
        true, true, (short) 1);
    currentExpectedOffset += us + ps; // skip one put
    verifyIndexEntry(indexEntries.get(4), (MockId) p2.getKey(), currentExpectedOffset, ps, expirationTime, 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, us, Utils.Infinite_Time, false,
        true, true, (short) 1);
    currentExpectedOffset += us + ps; // skip one put
    verifyIndexEntry(indexEntries.get(3), (MockId) p3.getKey(), currentExpectedOffset, ps, Utils.Infinite_Time, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(4), (MockId) p3.getKey(), currentExpectedOffset, us, Utils.Infinite_Time, false,
        false, true, (short) 1);
    currentExpectedOffset += us;

    // 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), (MockId) p3.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 1);
    currentExpectedOffset += ts + ps; // skip one put
    verifyIndexEntry(indexEntries.get(2), (MockId) p4.getKey(), currentExpectedOffset, ps, expirationTime, false, false,
        false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(3), (MockId) p4.getKey(), currentExpectedOffset, us, expirationTime, false, false,
        true, (short) 1);
    currentExpectedOffset += us;
    verifyIndexEntry(indexEntries.get(4), (MockId) p4.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 1);
    currentExpectedOffset += ts;

    // 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);
    currentExpectedOffset += ps; // skip one put
    verifyIndexEntry(indexEntries.get(1), (MockId) p5.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        true, false, (short) 1);
    currentExpectedOffset += ds;
    verifyIndexEntry(indexEntries.get(2), (MockId) p6.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        true, false, (short) 1);
    currentExpectedOffset += ds;
    verifyIndexEntry(indexEntries.get(3), (MockId) p7.getKey(), currentExpectedOffset, ps, Utils.Infinite_Time, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(4), (MockId) p7.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 0);
    currentExpectedOffset += ts;

    // 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);
    currentExpectedOffset += 3 * ps; // skip three puts
    verifyIndexEntry(indexEntries.get(3), (MockId) p8.getKey(), currentExpectedOffset, ps, expirationTime, false, false,
        false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(4), (MockId) p8.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 0);
    currentExpectedOffset += ts;

    // 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);
    currentExpectedOffset += 3 * ps; // skip three puts
    verifyIndexEntry(indexEntries.get(3), (MockId) p9.getKey(), currentExpectedOffset, ps, Utils.Infinite_Time, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(4), (MockId) p9.getKey(), currentExpectedOffset, us, Utils.Infinite_Time, false,
        false, true, (short) 2);
    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);
    verifyIndexEntry(indexEntries.get(0), (MockId) p9.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, 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);

    state.reloadIndex(true, false);
    logSegmentName = p11.getValue().getOffset().getName();
    segmentsUnderCompaction = Collections.singletonList(logSegmentName);
    deleteReferenceTimeMs = state.time.milliseconds();
    endOffsetOfSegmentBeforeCompaction = state.log.getSegment(logSegmentName).getEndOffset();
    details = new CompactionDetails(deleteReferenceTimeMs, 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
    // P11 U11 U11 T11 P| P12 T12 U12 U12 P| P13 T13 U13 U13 P| P14 T14 D14 D14 P| P15 T15 D15 D15 P | T16 P P P P
    // After compaction, the records in the log are
    // P11 U11 T11 P P12| T12 U12 P P13 T13| U13 P D14 P D15| P P P P P
    cleanedUpSize = us + us + us + (ps + ts + ds) + (ps + ts + ds) + ts;
    compactedLogSegmentName = logSegmentName.getNextGenerationName();
    assertEquals("End offset of log segment not as expected after compaction",
        endOffsetOfSegmentBeforeCompaction - cleanedUpSize,
        state.log.getSegment(compactedLogSegmentName).getEndOffset());

    compactedLogSegment = state.log.getSegment(compactedLogSegmentName);
    indexSegments = state.index.getIndexSegments()
        .subMap(new Offset(compactedLogSegmentName, 0), true,
            new Offset(compactedLogSegmentName.getNextPositionName(), 0), false);
    condition = new FindEntriesCondition(Long.MAX_VALUE);
    currentExpectedOffset = compactedLogSegment.getStartOffset();

    // Get the entries in the first segment
    segment = indexSegments.get(new Offset(compactedLogSegmentName, compactedLogSegment.getStartOffset()));
    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) p11.getKey(), currentExpectedOffset, ps, expirationTime, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(1), (MockId) p11.getKey(), currentExpectedOffset, us, expirationTime, false,
        false, true, (short) 2);
    currentExpectedOffset += us;
    verifyIndexEntry(indexEntries.get(2), (MockId) p11.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 2);
    currentExpectedOffset += ts + ps; // skip one put
    verifyIndexEntry(indexEntries.get(4), (MockId) p12.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) p12.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 0);
    currentExpectedOffset += ts;
    verifyIndexEntry(indexEntries.get(1), (MockId) p12.getKey(), currentExpectedOffset, us, Utils.Infinite_Time, false,
        true, true, (short) 2);
    currentExpectedOffset += us + ps; // skip one put
    verifyIndexEntry(indexEntries.get(3), (MockId) p13.getKey(), currentExpectedOffset, ps, expirationTime, false,
        false, false, (short) 0);
    currentExpectedOffset += ps;
    verifyIndexEntry(indexEntries.get(4), (MockId) p13.getKey(), currentExpectedOffset, ts, Utils.Infinite_Time, false,
        true, false, (short) 0);
    currentExpectedOffset += ts;

    // 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), (MockId) p13.getKey(), currentExpectedOffset, us, Utils.Infinite_Time, false,
        true, true, (short) 2);
    currentExpectedOffset += us + ps; // skip one put
    verifyIndexEntry(indexEntries.get(2), (MockId) p14.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        true, false, (short) 1);
    currentExpectedOffset += us + ps; // skip one put
    verifyIndexEntry(indexEntries.get(4), (MockId) p15.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        true, false, (short) 1);
    currentExpectedOffset += ds;

    // 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);
  }
Tests when the PUT, DELETE, TTL_UPDATE AND UNDELETE of the same key are in the same index segment.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java