@Test
  public void findEntriesSinceIncarnationIdTest() throws StoreException {
    Offset lastRecordOffset = state.index.journal.getLastOffset();
    state.appendToLog(2 * CuratedLogIndexState.PUT_RECORD_SIZE);
    // will be recovered
    FileSpan firstRecordFileSpan =
        state.log.getFileSpanForMessage(state.index.getCurrentEndOffset(), CuratedLogIndexState.PUT_RECORD_SIZE);
    // will not be recovered
    FileSpan secondRecordFileSpan =
        state.log.getFileSpanForMessage(firstRecordFileSpan.getEndOffset(), CuratedLogIndexState.PUT_RECORD_SIZE);

    UUID oldSessionId = state.sessionId;
    UUID oldIncarnationId = state.incarnationId;
    final MockId newId = state.getUniqueId();
    short accountId = Utils.getRandomShort(TestUtils.RANDOM);
    short containerId = Utils.getRandomShort(TestUtils.RANDOM);
    long operationTimeMs = state.time.milliseconds();

    // add to allKeys() so that doFindEntriesSinceTest() works correctly.
    IndexValue putValue =
        new IndexValue(CuratedLogIndexState.PUT_RECORD_SIZE, firstRecordFileSpan.getStartOffset(), Utils.Infinite_Time,
            operationTimeMs, accountId, containerId);
    state.allKeys.computeIfAbsent(newId, k -> new TreeSet<>()).add(putValue);
    state.recovery = (read, startOffset, endOffset, factory) -> new MessageStoreRecovery.RecoveryResult(
        Collections.singletonList(
            new MessageInfo(newId, CuratedLogIndexState.PUT_RECORD_SIZE, accountId, containerId, operationTimeMs)),
        null, endOffset);
    // change in incarnationId
    state.incarnationId = UUID.randomUUID();
    state.reloadIndex(true, true);

    long bytesRead = state.index.getAbsolutePositionInLogForOffset(firstRecordFileSpan.getEndOffset());
    // create a token that will be past the index end offset on startup after recovery with old incarnationId
    StoreFindToken startToken =
        new StoreFindToken(secondRecordFileSpan.getEndOffset(), oldSessionId, oldIncarnationId, false, null, null,
            UNINITIALIZED_RESET_KEY_VERSION);
    // token should get reset internally, all keys should be returned and the returned token should be pointing to
    // start offset of firstRecordFileSpan.
    IndexSegment segmentOfToken =
        state.index.getIndexSegments().floorEntry(firstRecordFileSpan.getStartOffset()).getValue();
    StoreFindToken expectedEndToken =
        new StoreFindToken(firstRecordFileSpan.getStartOffset(), state.sessionId, state.incarnationId, false,
            segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
    expectedEndToken.setBytesRead(bytesRead);
    doFindEntriesSinceTest(startToken, Long.MAX_VALUE, state.allKeys.keySet(), expectedEndToken);

    // create a token that is not past the index end offset on startup after recovery with old incarnationId.
    // token should get reset internally, all keys should be returned and the returned token should be be pointing to
    // start offset of firstRecordFileSpan.
    startToken = new StoreFindToken(lastRecordOffset, oldSessionId, oldIncarnationId, false, null, null,
        UNINITIALIZED_RESET_KEY_VERSION);
    expectedEndToken =
        new StoreFindToken(firstRecordFileSpan.getStartOffset(), state.sessionId, state.incarnationId, false,
            segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
    expectedEndToken.setBytesRead(bytesRead);
    doFindEntriesSinceTest(startToken, Long.MAX_VALUE, state.allKeys.keySet(), expectedEndToken);
  }
Tests behaviour of {@link PersistentIndex#findEntriesSince(FindToken, long)} relating to incarnationId
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java