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

    // if there is no bad shutdown but the store token is past the index end offset, it is an error state
    StoreFindToken startToken =
        new StoreFindToken(secondRecordFileSpan.getStartOffset(), new UUID(1, 1), state.incarnationId, false, null,
            null, UNINITIALIZED_RESET_KEY_VERSION);
    doFindEntriesSinceFailureTest(startToken, StoreErrorCodes.UnknownError);

    UUID oldSessionId = state.sessionId;
    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);
    state.reloadIndex(true, true);

    // If there is no incarnationId in the incoming token, for backwards compatibility purposes we consider it as valid
    // and proceed with session id validation and so on.
    UUID[] incarnationIds = new UUID[]{state.incarnationId, null};
    for (UUID incarnationIdToTest : incarnationIds) {
      long bytesRead = state.index.getAbsolutePositionInLogForOffset(firstRecordFileSpan.getEndOffset());
      // create a token that will be past the index end offset on startup after recovery.
      if (incarnationIdToTest == null) {
        startToken = getTokenWithNullIncarnationId(
            new StoreFindToken(secondRecordFileSpan.getEndOffset(), oldSessionId, state.incarnationId, false, null,
                null, UNINITIALIZED_RESET_KEY_VERSION));
        assertNull("IncarnationId is expected to be null ", startToken.getIncarnationId());
      } else {
        startToken =
            new StoreFindToken(secondRecordFileSpan.getEndOffset(), oldSessionId, incarnationIdToTest, false, null,
                null, UNINITIALIZED_RESET_KEY_VERSION);
      }
      // token should get reset internally, no keys should be returned and the returned token should be correct (offset
      // in it will be the current log end offset = firstRecordFileSpan.getEndOffset()). The returned token should have correct reset key info.
      IndexSegment segmentOfToken =
          state.index.getIndexSegments().floorEntry(firstRecordFileSpan.getEndOffset()).getValue();
      StoreFindToken expectedEndToken =
          new StoreFindToken(firstRecordFileSpan.getEndOffset(), state.sessionId, state.incarnationId, true,
              segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
      expectedEndToken.setBytesRead(bytesRead);
      doFindEntriesSinceTest(startToken, Long.MAX_VALUE, Collections.emptySet(), expectedEndToken);

      // create a token that is not past the index end offset on startup after recovery. Should work as expected
      if (incarnationIdToTest == null) {
        startToken = getTokenWithNullIncarnationId(
            new StoreFindToken(lastRecordOffset, oldSessionId, state.incarnationId, false, null, null,
                UNINITIALIZED_RESET_KEY_VERSION));
        assertNull("IncarnationId is expected to be null ", startToken.getIncarnationId());
      } else {
        startToken = new StoreFindToken(lastRecordOffset, oldSessionId, incarnationIdToTest, false, null, null,
            UNINITIALIZED_RESET_KEY_VERSION);
      }
      segmentOfToken = state.index.getIndexSegments().floorEntry(firstRecordFileSpan.getStartOffset()).getValue();
      expectedEndToken =
          new StoreFindToken(firstRecordFileSpan.getStartOffset(), state.sessionId, state.incarnationId, false,
              segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
      expectedEndToken.setBytesRead(bytesRead);
      doFindEntriesSinceTest(startToken, Long.MAX_VALUE, Collections.singleton(newId), expectedEndToken);
    }
  }
Tests behaviour of {@link PersistentIndex#findEntriesSince(FindToken, long)} on crash-restart of index and some
recovery. Specifically tests cases where tokens have been handed out before the "crash" failure.
@throws IOException
@throws StoreException
ambry-store/src/test/java/com/github/ambry/store/IndexTest.java