@Test
  public void testBeforeAndAfterOffsetSanityCheck() throws Exception {
    assumeTrue(isLogSegmented && persistentIndexVersion >= PersistentIndex.VERSION_3);
    // Adding valid before and after offsets for finished compaction
    Map<Offset, Offset> beforeAndAfter = new HashMap<>();
    Offset after = state.index.getIndexSegments().firstKey();
    Offset before = new Offset(after.getName().getNextGenerationName(), LogSegment.HEADER_SIZE);
    beforeAndAfter.put(before, after); // This is one redirection.
    Offset beforeBefore = new Offset(before.getName().getNextGenerationName(), LogSegment.HEADER_SIZE);
    beforeAndAfter.put(beforeBefore, before); // This is two redirections.
    state.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(state.time.milliseconds(), beforeAndAfter, false);
    state.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();
    assertFalse(state.index.isSanityCheckFailed());

    // Adding invalid before and after offsets for inprogress compaction
    try {
      state.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(state.time.milliseconds(), beforeAndAfter, true);
      fail("Before and after offsets should be invalid for inprogress compaction");
    } catch (Exception e) {
    }

    // Adding invalid before and after offset for finished compaction.
    beforeAndAfter.clear();
    Offset secondIndexSegmentOffset = state.index.getIndexSegments().higherKey(after);
    // Before offset exist in the current index
    beforeAndAfter.put(secondIndexSegmentOffset, after);
    state.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(state.time.milliseconds(), beforeAndAfter, false);
    state.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();

    assertTrue(state.index.isSanityCheckFailed());

    state.reloadIndex(true, false);
    assertFalse(state.index.isSanityCheckFailed());
    beforeAndAfter.clear();
    // Create two non-existing offsets
    Offset nonExistingBefore =
        new Offset(secondIndexSegmentOffset.getName().getNextGenerationName(), LogSegment.HEADER_SIZE);
    Offset nonExistingAfter = new Offset(nonExistingBefore.getName().getNextGenerationName(), LogSegment.HEADER_SIZE);
    beforeAndAfter.put(nonExistingBefore, nonExistingAfter);
    state.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(state.time.milliseconds(), beforeAndAfter, false);
    state.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();
    assertTrue(state.index.isSanityCheckFailed());

    state.reloadIndex(true, false);
    assertFalse(state.index.isSanityCheckFailed());
    beforeAndAfter.clear();
    // Create a loop
    beforeAndAfter.put(nonExistingBefore, nonExistingAfter);
    beforeAndAfter.put(nonExistingAfter, nonExistingBefore);
    state.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(state.time.milliseconds(), beforeAndAfter, false);
    state.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();
    assertTrue(state.index.isSanityCheckFailed());
  }
Test the sanity check of the before and after index segment offsets.
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java