@Test
  public void emptyOneLogSegmentInCompactionTest() throws Exception {
    refreshState(false, false, false);
    // Set up log segment like this
    // 0_0 doesn't have any valid values (all expired put), will compact 0_0 by itself.
    // 1_0 is full of PUT, some of them are going to be compacted.
    // 2_0 doesn't have any valid values (all expired put), will compact 1_0 and 2_0 together
    // 3_0 is full of valid PUTs
    // 4_0 is doesn't have any valid values (all expired put), will compact 4_0 by itself.

    int numPuts = (int) ((state.log.getSegmentCapacity() - LogSegment.HEADER_SIZE) / (PUT_RECORD_SIZE));
    state.addPutEntries(numPuts, PUT_RECORD_SIZE, 0);
    assertEquals(1, state.index.getLogSegmentCount());
    state.addPutEntries(numPuts - 3, PUT_RECORD_SIZE, Utils.Infinite_Time);
    state.addPutEntries(3, PUT_RECORD_SIZE, 0);
    assertEquals(2, state.index.getLogSegmentCount());
    state.addPutEntries(numPuts, PUT_RECORD_SIZE, 0);
    assertEquals(3, state.index.getLogSegmentCount());
    state.addPutEntries(numPuts, PUT_RECORD_SIZE, Utils.Infinite_Time);
    assertEquals(4, state.index.getLogSegmentCount());
    state.addPutEntries(numPuts, PUT_RECORD_SIZE, 0);
    assertEquals(5, state.index.getLogSegmentCount());
    writeDataToMeetRequiredSegmentCount(6, Collections.singletonList(Utils.Infinite_Time));

    state.time.setCurrentMilliseconds(System.currentTimeMillis());
    state.reloadIndex(true, false);

    // Compact 0_0, the target would be empty, and there is no index segment before the target, the before and after index
    // segment offset map would be empty in compaction log.
    List<LogSegmentName> segmentsUnderCompaction = getLogSegments(0, 1);
    compactor = getCompactor(state.log, DISK_IO_SCHEDULER, null, false);
    compactor.initialize(state.index);

    CompactionDetails details = new CompactionDetails(time.milliseconds(), segmentsUnderCompaction, null);
    try {
      compactor.compact(details, bundleReadBuffer);
    } finally {
      compactor.close(0);
    }

    CompactionLog latestLog = getLatestCompactionLog();
    assertEquals(0, latestLog.getBeforeAndAfterIndexSegmentOffsets().size());

    state.reloadIndex(true, false);
    // 0_0 is gone, now 1_0 is the first log segment
    segmentsUnderCompaction = getLogSegments(0, 2);
    details = new CompactionDetails(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);
    }
    state.reloadIndex(true, false);
    latestLog = getLatestCompactionLog();
    Offset lastIndexSegmentOffsetForTarget =
        state.index.getIndexSegments().floorKey(new Offset(LogSegmentName.fromPositionAndGeneration(3, 0), 0));
    assertEquals(LogSegmentName.fromPositionAndGeneration(1, 1), lastIndexSegmentOffsetForTarget.getName());
    for (Map.Entry<Offset, Offset> entry : latestLog.getBeforeAndAfterIndexSegmentOffsets().entrySet()) {
      if (entry.getKey().getName().equals(LogSegmentName.fromPositionAndGeneration(2, 0))) {
        assertEquals(lastIndexSegmentOffsetForTarget, entry.getValue());
      }
    }

    state.reloadIndex(true, false);
    // 0_0 and 2_0 are gone, now it's 1_1, 3_0, 4_0
    segmentsUnderCompaction = getLogSegments(2, 1);
    details = new CompactionDetails(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);
    }
    state.reloadIndex(true, false);
    latestLog = getLatestCompactionLog();
    Offset lastIndexSegmentOffsetFromSource =
        state.index.getIndexSegments().floorKey(new Offset(LogSegmentName.fromPositionAndGeneration(4, 0), 0));
    assertEquals(LogSegmentName.fromPositionAndGeneration(3, 0), lastIndexSegmentOffsetFromSource.getName());
    for (Map.Entry<Offset, Offset> entry : latestLog.getBeforeAndAfterIndexSegmentOffsets().entrySet()) {
      assertEquals(lastIndexSegmentOffsetFromSource, entry.getValue());
    }
  }
Test the cases where the whole log segment is removed by the compaction, before and after compaction index segment
offsets has to be correct.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/BlobStoreCompactorTest.java