@Test
  public void orphanedPutRecordsTest() throws Exception {
    refreshState(false, false, false);
    // write a PUT record that will be "lost"
    MockId orphanedId = (MockId) state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time).get(0).getKey();
    // add a delete entry for  orphanedId
    state.addDeleteEntry(orphanedId);
    // get the index value and "lose" the PUT record. This works because we get a reference to the value in the index.
    IndexValue value = state.index.findKey(orphanedId);
    value.clearOriginalMessageOffset();
    // add a put entry that spans the rest of the log segment
    long lastRecSize = state.log.getSegmentCapacity() - state.index.getCurrentEndOffset().getOffset();
    state.addPutEntries(1, lastRecSize, Utils.Infinite_Time);
    // add an entry so that a new log segment is created
    state.addPutEntries(1, PUT_RECORD_SIZE, Utils.Infinite_Time);

    state.reloadIndex(true, false);
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 1);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), true);

    // make sure the delete still exists
    checkIndexValue(orphanedId);
    // the first log segment should not contain the the PUT that had no index entry
    LogSegment firstLogSegment = state.log.getFirstSegment();
    long size = firstLogSegment.getEndOffset() - firstLogSegment.getStartOffset();
    assertEquals("Segment size not as expected", lastRecSize + DELETE_RECORD_SIZE, size);
  }
Tests compaction on a log that contains a PUT record that has no corresponding entry in the index (this can happen
due to recovery corner cases - refer to recovery code in PersistentIndex.java).
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java