@Test
  public void rebuildTokenBasedOnCompactionHistoryTest() throws StoreException {
    assumeTrue(isLogSegmented);
    stateForTokenTest.index.enableRebuildTokenBasedOnCompactionHistory();

    IndexSegment firstIndexSegment = stateForTokenTest.index.getIndexSegments().firstEntry().getValue();
    IndexSegment secondIndexSegment =
        stateForTokenTest.index.getIndexSegments().higherEntry(firstIndexSegment.getStartOffset()).getValue();
    long now = stateForTokenTest.time.milliseconds();
    Map<Offset, Offset> indexSegmentOffsets = new HashMap<>();
    Offset secondIndexSegmentOffset = secondIndexSegment.getStartOffset();
    Offset before =
        new Offset(secondIndexSegmentOffset.getName().getNextGenerationName(), secondIndexSegmentOffset.getOffset());
    indexSegmentOffsets.put(before, secondIndexSegmentOffset);
    stateForTokenTest.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(now, indexSegmentOffsets, false);
    stateForTokenTest.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();

    // 1. Generate an invalid index-based token with invalid offset, but it can be found in the before and after compaction map
    Offset invalidOffset = before;
    StoreKey keyFromFirstIndexSegment = firstIndexSegment.iterator().next().getKey();
    StoreKey keyFromSecondIndexSegment = secondIndexSegment.iterator().next().getKey();
    StoreFindToken startToken = new StoreFindToken(keyFromFirstIndexSegment, invalidOffset, stateForTokenTest.sessionId,
        stateForTokenTest.incarnationId, null, null, (short) -1);
    // The invalid token should be revalidated by looking up the before and after compaction index segment map
    FindInfo findInfo = stateForTokenTest.index.findEntriesSince(startToken, 1);
    StoreFindToken token = (StoreFindToken) findInfo.getFindToken();
    StoreFindToken expectedToken =
        new StoreFindToken(keyFromSecondIndexSegment, secondIndexSegmentOffset, stateForTokenTest.sessionId,
            stateForTokenTest.incarnationId, secondIndexSegment.getResetKey(), secondIndexSegment.getResetKeyType(),
            secondIndexSegment.getResetKeyLifeVersion());
    compareTokens(expectedToken, token);

    // 2. Generate an invalid index-based token with invalid offset, but it can't be found in the before and after compaction map
    invalidOffset = new Offset(before.getName().getNextGenerationName(), before.getOffset());
    startToken = new StoreFindToken(keyFromSecondIndexSegment, invalidOffset, stateForTokenTest.sessionId,
        stateForTokenTest.incarnationId, null, null, (short) -1);
    // The invalid token will be revalidated to the beginning of the log
    findInfo = stateForTokenTest.index.findEntriesSince(startToken, 1);
    token = (StoreFindToken) findInfo.getFindToken();
    expectedToken =
        new StoreFindToken(keyFromFirstIndexSegment, firstIndexSegment.getStartOffset(), stateForTokenTest.sessionId,
            stateForTokenTest.incarnationId, firstIndexSegment.getResetKey(), firstIndexSegment.getResetKeyType(),
            firstIndexSegment.getResetKeyLifeVersion());
    compareTokens(expectedToken, token);

    // 3. Generate an invalid index-based token with invalid offset, but it can be found by multiple redirections.
    Offset beforeBefore = new Offset(before.getName().getNextGenerationName(), before.getOffset() + 100);
    indexSegmentOffsets.put(beforeBefore, before);
    stateForTokenTest.index.updateBeforeAndAfterCompactionIndexSegmentOffsets(now, indexSegmentOffsets, false);
    stateForTokenTest.index.sanityCheckBeforeAndAfterCompactionIndexSegmentOffsets();
    invalidOffset = beforeBefore;
    startToken = new StoreFindToken(keyFromFirstIndexSegment, invalidOffset, stateForTokenTest.sessionId,
        stateForTokenTest.incarnationId, null, null, (short) -1);
    // The invalid token should be revalidated by looking up the before and after compaction index segment map
    findInfo = stateForTokenTest.index.findEntriesSince(startToken, 1);
    token = (StoreFindToken) findInfo.getFindToken();
    expectedToken = new StoreFindToken(keyFromSecondIndexSegment, secondIndexSegmentOffset, stateForTokenTest.sessionId,
        stateForTokenTest.incarnationId, secondIndexSegment.getResetKey(), secondIndexSegment.getResetKeyType(),
        secondIndexSegment.getResetKeyLifeVersion());
    compareTokens(expectedToken, token);

    // Generate an invalid journal-based token with invalid offset, but it can be found in the before and after compaction map
    invalidOffset = new Offset(before.getName(), before.getOffset() + 100);
    startToken =
        new StoreFindToken(invalidOffset, stateForTokenTest.sessionId, stateForTokenTest.incarnationId, false, null,
            null, (short) -1);
    // The invalid token should be revalidated by looking up the before and after compaction index segment map
    findInfo = stateForTokenTest.index.findEntriesSince(startToken, 1);
    token = (StoreFindToken) findInfo.getFindToken();
    expectedToken = new StoreFindToken(keyFromSecondIndexSegment, secondIndexSegmentOffset, stateForTokenTest.sessionId,
        stateForTokenTest.incarnationId, secondIndexSegment.getResetKey(), secondIndexSegment.getResetKeyType(),
        secondIndexSegment.getResetKeyLifeVersion());
    compareTokens(expectedToken, token);
  }
Tests cases where find token is rebuilt based on reset key.
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java