@Test
  public void deleteWithOlderVersionTest() throws Exception {
    assumeTrue(withUndelete);
    // 1. Compact all the puts and deletes with delete reference time being 0
    List<IndexEntry> entries = createDeleteWithOlderVersion();
    int numPuts = entries.size();
    long logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
    long realBytesBeforeCompaction =
        logSegmentSizeSumBeforeCompaction - LogSegment.HEADER_SIZE * state.index.getLogSegmentCount();
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    compactAndVerify(segmentsUnderCompaction, 0, true);
    long compactedBytes = numPuts * (DELETE_RECORD_SIZE + UNDELETE_RECORD_SIZE);
    long logSegmentSizeSumAfterCompaction = getSumOfLogSegmentEndOffsets();
    long realBytesAfterCompaction =
        logSegmentSizeSumAfterCompaction - LogSegment.HEADER_SIZE * state.index.getLogSegmentCount();
    assertEquals(compactedBytes, realBytesBeforeCompaction - realBytesAfterCompaction);

    for (IndexEntry entry : entries) {
      List<IndexValue> values = state.index.findAllIndexValuesForKey(entry.getKey(), null);
      assertNotNull(values);
      assertEquals(2, values.size());
      assertEquals((short) 1, values.get(0).getLifeVersion());
      assertTrue(values.get(0).isDelete());
      assertTrue(values.get(1).isPut());
    }

    // 2. Compact all the puts and deletes with delete reference time being now.
    entries = createDeleteWithOlderVersion();
    numPuts = entries.size();
    logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
    realBytesBeforeCompaction =
        logSegmentSizeSumBeforeCompaction - LogSegment.HEADER_SIZE * state.index.getLogSegmentCount();
    segmentsUnderCompaction = getLogSegments(0, state.index.getLogSegmentCount() - 1);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), true);
    compactedBytes = numPuts * (DELETE_RECORD_SIZE + UNDELETE_RECORD_SIZE + PUT_RECORD_SIZE);
    logSegmentSizeSumAfterCompaction = getSumOfLogSegmentEndOffsets();
    realBytesAfterCompaction =
        logSegmentSizeSumAfterCompaction - LogSegment.HEADER_SIZE * state.index.getLogSegmentCount();
    assertEquals(compactedBytes, realBytesBeforeCompaction - realBytesAfterCompaction);

    for (IndexEntry entry : entries) {
      List<IndexValue> values = state.index.findAllIndexValuesForKey(entry.getKey(), null);
      assertNotNull(values);
      assertEquals(1, values.size());
      assertEquals((short) 1, values.get(0).getLifeVersion());
      assertTrue(values.get(0).isDelete());
    }

    // 3. Compact only the deletes, with delete reference time being now.
    entries = createDeleteWithOlderVersion();
    numPuts = entries.size();
    logSegmentSizeSumBeforeCompaction = getSumOfLogSegmentEndOffsets();
    realBytesBeforeCompaction =
        logSegmentSizeSumBeforeCompaction - LogSegment.HEADER_SIZE * state.index.getLogSegmentCount();
    segmentsUnderCompaction = getLogSegments(1, state.index.getLogSegmentCount() - 2);
    compactAndVerify(segmentsUnderCompaction, state.time.milliseconds(), true);
    // there is one put at the beginning of the second log segment.
    compactedBytes = numPuts * (DELETE_RECORD_SIZE + UNDELETE_RECORD_SIZE) + PUT_RECORD_SIZE;
    logSegmentSizeSumAfterCompaction = getSumOfLogSegmentEndOffsets();
    realBytesAfterCompaction =
        logSegmentSizeSumAfterCompaction - LogSegment.HEADER_SIZE * state.index.getLogSegmentCount();
    assertEquals(compactedBytes, realBytesBeforeCompaction - realBytesAfterCompaction);
  }
Test the case when there are multiple deletes in the log.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java