@Test
  public void findEntriesSinceOnCrashRestartTest() throws StoreException {
    UUID oldSessionId = state.sessionId;
    state.addPutEntries(3, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    state.reloadIndex(true, true);
    // create a token that will be past the log end offset on start up after restart.
    StoreFindToken startToken = new StoreFindToken(new Offset(state.log.getEndOffset().getName(),
        (state.log.getEndOffset().getOffset() + (2 * CuratedLogIndexState.PUT_RECORD_SIZE))), oldSessionId,
        state.incarnationId, false, null, null, UNINITIALIZED_RESET_KEY_VERSION);
    // end token should point to log end offset on startup
    long bytesRead = state.index.getAbsolutePositionInLogForOffset(state.log.getEndOffset());
    IndexSegment segmentOfToken = state.index.getIndexSegments().floorEntry(state.log.getEndOffset()).getValue();
    StoreFindToken expectedEndToken =
        new StoreFindToken(state.log.getEndOffset(), state.sessionId, state.incarnationId, true,
            segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
    expectedEndToken.setBytesRead(bytesRead);
    // Fetch the FindToken returned from findEntriesSince
    FindInfo findInfo = state.index.findEntriesSince(startToken, Long.MAX_VALUE);
    assertEquals("EndToken mismatch ", expectedEndToken, findInfo.getFindToken());
    assertEquals("No entries should have been returned ", 0, findInfo.getMessageEntries().size());

    // add 2 entries to index and log
    Set<MockId> expectedEntries = new HashSet<>();
    state.addPutEntries(1, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    expectedEntries.add(state.logOrder.lastEntry().getValue().getFirst());
    state.addPutEntries(1, CuratedLogIndexState.PUT_RECORD_SIZE, Utils.Infinite_Time);
    expectedEntries.add(state.logOrder.lastEntry().getValue().getFirst());
    bytesRead = state.index.getAbsolutePositionInLogForOffset(state.index.getCurrentEndOffset());
    segmentOfToken = state.index.getIndexSegments().floorEntry(state.index.journal.getLastOffset()).getValue();
    expectedEndToken =
        new StoreFindToken(state.index.journal.getLastOffset(), state.sessionId, state.incarnationId, false,
            segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
    expectedEndToken.setBytesRead(bytesRead);
    doFindEntriesSinceTest((StoreFindToken) findInfo.getFindToken(), Long.MAX_VALUE, expectedEntries, expectedEndToken);
  }
Tests behaviour of {@link PersistentIndex#findEntriesSince(FindToken, long)} on crash restart for the following
scenario
After restart, lets say no new writes have gone into the store.
For a findEntriesSince(offset beyond logEndOffsetOnStartup with different session id) call, index will reset the
token to logEndOffsetOnStartup and returns the same.
On the subsequent findEntriesSince() call, the index should start returning entries
starting from that offset and should not consider that token as non-inclusive
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java