@Test
  public void undeleteSameIndexSegmentTest_NoTtlUpdate() 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() + 200 * Time.MsPerSec;
    // Index Segment 0.1 P D U -> P U
    IndexEntry p1 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.addDeleteEntry((MockId) p1.getKey());
    state.addUndeleteEntry((MockId) p1.getKey());
    state.addPutEntries(2, PUT_RECORD_SIZE, Utils.Infinite_Time);

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

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

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

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

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

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

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

    // Index Segment 0.9 P U -> P U
    IndexEntry p9 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0);
    state.addUndeleteEntry((MockId) p9.getKey(), (short) 2);
    state.addPutEntries(3, 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 -> []
    IndexEntry p11 = state.addPutEntries(1, PUT_RECORD_SIZE, expirationTime).get(0);
    state.addUndeleteEntry((MockId) p11.getKey(), (short) 2);
    state.addPutEntries(3, PUT_RECORD_SIZE, Utils.Infinite_Time);

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

    // Index Segment 1.3 P2 -> P2
    IndexEntry p13 = state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time, (short) 2).get(0);
    state.addPutEntries(4, PUT_RECORD_SIZE, Utils.Infinite_Time);

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

    final LogSegmentName firstLogSegmentName = p1.getValue().getOffset().getName();
    TreeSet<Offset> firstLogSegmentIndexOffsets = state.index.getIndexSegments()
        .keySet()
        .stream()
        .filter(offset -> offset.getName().equals(firstLogSegmentName))
        .collect(Collectors.toCollection(TreeSet::new));
    final LogSegmentName secondLogSegmentName = firstLogSegmentName.getNextPositionName();
    TreeSet<Offset> secondLogSegmentIndexOffsets = state.index.getIndexSegments()
        .keySet()
        .stream()
        .filter(offset -> offset.getName().equals(secondLogSegmentName))
        .collect(Collectors.toCollection(TreeSet::new));

    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 D1 U1 P P| P2 D2 U2 P P| P3 D3 U3 D3 U3| P4 D4 U4 D4 U4| P5 D5 D5 U5 P| P6 D6 U6 D6 P| P7 D7 U7 D7 P |P8 D8 D8 P P| P9 U9 P P P
    // After compaction, the records in the log are
    // P1 U1 P P P| P P3 U3 P5 U5| P D6 P D7 P| D8 P P P9 U9| P 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 =
        ds + (ps + ds + us) + (2 * ds + us) + (2 * ds + 2 * us + ps) + 2 * ds + (ps + ds + us) + (ps + ds + us) + (ps
            + ds) + 0;
    LogSegmentName compactedLogSegmentName = logSegmentName.getNextGenerationName();
    assertEquals("End offset of log segment not as expected after compaction",
        endOffsetOfSegmentBeforeCompaction - cleanedUpSize,
        state.log.getSegment(compactedLogSegmentName).getEndOffset());

    CompactionLog cLog = getLatestCompactionLog();
    Map<Offset, Offset> indexSegmentOffsets = cLog.getBeforeAndAfterIndexSegmentOffsets();
    Map<Offset, Offset> expectedIndexSegmentOffsets = new HashMap<>();

    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, us, Utils.Infinite_Time, false,
        false, true, (short) 1);
    currentExpectedOffset += us + 3 * ps; // skip three puts

    Offset before = firstLogSegmentIndexOffsets.first();
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 1 -> after 1
    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 2 -> after 1

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

    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 3 -> after 2
    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 4 -> after 2
    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 5 -> after 2

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

    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 6 -> after 3
    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 7 -> after 3
    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 8 -> after 3
    // when copying 8th index segment, the third index segment after compaction didn't roll over yet.

    // 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), (MockId) p8.getKey(), currentExpectedOffset, ds, Utils.Infinite_Time, true,
        false, false, (short) 3);
    currentExpectedOffset += ds + 2 * ps;
    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);

    before = firstLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 9 -> after 4

    for (Map.Entry<Offset, Offset> expectedEntry : expectedIndexSegmentOffsets.entrySet()) {
      Offset expectedBefore = expectedEntry.getKey();
      Offset expectedAfter = expectedEntry.getValue();
      Offset obtainedAfter = indexSegmentOffsets.get(expectedBefore);
      assertEquals("Mismatch for offset " + expectedBefore + " " + firstLogSegmentIndexOffsets, expectedAfter,
          obtainedAfter);
    }
    expectedIndexSegmentOffsets.clear();

    // 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 P P P| P12 D12 U12 U12 P | P13(2) P P P P
    // After compaction, the records in the log are
    // P P P P12 U12| P P13(2)
    cleanedUpSize = (ps + us) + (ds + us);
    compactedLogSegmentName = logSegmentName.getNextGenerationName();
    assertEquals("End offset of log segment not as expected after compaction",
        endOffsetOfSegmentBeforeCompaction - cleanedUpSize,
        state.log.getSegment(compactedLogSegmentName).getEndOffset());

    cLog = getLatestCompactionLog();
    indexSegmentOffsets = cLog.getBeforeAndAfterIndexSegmentOffsets();

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

    before = secondLogSegmentIndexOffsets.first();
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 1 -> after 1
    before = secondLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 2 -> after 1

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

    before = secondLogSegmentIndexOffsets.higher(before);
    expectedIndexSegmentOffsets.put(before, segment.getStartOffset()); // before 3 -> after 2

    for (Map.Entry<Offset, Offset> expectedEntry : expectedIndexSegmentOffsets.entrySet()) {
      Offset expectedBefore = expectedEntry.getKey();
      Offset expectedAfter = expectedEntry.getValue();
      Offset obtainedAfter = indexSegmentOffsets.get(expectedBefore);
      assertEquals("Mismatch for offset " + expectedBefore + " " + secondLogSegmentIndexOffsets, expectedAfter,
          obtainedAfter);
    }
    expectedIndexSegmentOffsets.clear();

    // 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 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