FindInfo findEntriesSince(FindToken token, long maxTotalSizeOfEntries) throws StoreException {
    long startTimeInMs = time.milliseconds();
    try {
      Offset logEndOffsetBeforeFind = log.getEndOffset();
      StoreFindToken storeToken = resetTokenIfRequired((StoreFindToken) token);
      logger.trace("Time used to validate token: {}", (time.milliseconds() - startTimeInMs));
      ConcurrentSkipListMap<Offset, IndexSegment> indexSegments = validIndexSegments;
      storeToken = revalidateStoreFindToken(storeToken, indexSegments, false);

      List<MessageInfo> messageEntries = new ArrayList<>();
      if (!storeToken.getType().equals(FindTokenType.IndexBased)) {
        startTimeInMs = time.milliseconds();
        Offset offsetToStart = storeToken.getOffset();
        if (storeToken.getType().equals(FindTokenType.Uninitialized)) {
          offsetToStart = getStartOffset();
        }
        logger.trace("Index : {} getting entries since {}", dataDir, offsetToStart);
        // check journal
        List<JournalEntry> entries = journal.getEntriesSince(offsetToStart, storeToken.getInclusive());
        logger.trace("Journal based token, Time used to get entries: {}", (time.milliseconds() - startTimeInMs));
        // we deliberately obtain a snapshot of the index segments AFTER fetching from the journal. This ensures that
        // any and all entries returned from the journal are guaranteed to be in the obtained snapshot of indexSegments.
        indexSegments = validIndexSegments;

        // recheck to make sure that offsetToStart hasn't slipped out of the journal. It is possible (but highly
        // improbable) that the offset has slipped out of the journal and has been compacted b/w the time of getting
        // entries from the journal to when another view of the index segments was obtained.
        Offset journalFirstOffset = journal.getFirstOffset();
        if (entries != null && journalFirstOffset != null && offsetToStart.compareTo(journalFirstOffset) >= 0) {
          startTimeInMs = time.milliseconds();
          logger.trace("Index : {} retrieving from journal from offset {} total entries {}", dataDir, offsetToStart,
              entries.size());
          Offset offsetEnd = offsetToStart;
          AtomicLong currentTotalSizeOfEntries = new AtomicLong(0);
          Offset endOffsetOfSnapshot = getCurrentEndOffset(indexSegments);
          Map<StoreKey, MessageInfo> messageInfoCache = new HashMap<>();
          for (JournalEntry entry : entries) {
            addMessageInfoForJournalEntry(entry, endOffsetOfSnapshot, indexSegments, messageInfoCache, messageEntries,
                currentTotalSizeOfEntries);
            offsetEnd = entry.getOffset();
            if (currentTotalSizeOfEntries.get() >= maxTotalSizeOfEntries) {
              break;
            }
          }
          logger.trace("Journal based token, Time used to generate message entries: {}",
              (time.milliseconds() - startTimeInMs));

          startTimeInMs = time.milliseconds();
          logger.trace("Index : {} new offset from find info {}", dataDir, offsetEnd);
          eliminateDuplicates(messageEntries);
          logger.trace("Journal based token, Time used to eliminate duplicates: {}",
              (time.milliseconds() - startTimeInMs));
          IndexSegment segmentOfToken = indexSegments.floorEntry(offsetEnd).getValue();
          StoreFindToken storeFindToken =
              new StoreFindToken(offsetEnd, sessionId, incarnationId, false, segmentOfToken.getResetKey(),
                  segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
          // use the latest ref of indexSegments
          long bytesRead = getTotalBytesRead(storeFindToken, messageEntries, logEndOffsetBeforeFind, indexSegments);
          storeFindToken.setBytesRead(bytesRead);
          return new FindInfo(messageEntries, storeFindToken);
        } else {
          storeToken = revalidateStoreFindToken(storeToken, indexSegments, false);
          offsetToStart = storeToken.getType().equals(FindTokenType.Uninitialized) ? getStartOffset(indexSegments)
              : storeToken.getOffset();
          // Find index segment closest to the token offset.
          // Get entries starting from the first key in this offset.
          Map.Entry<Offset, IndexSegment> entry = indexSegments.floorEntry(offsetToStart);
          StoreFindToken newToken;
          //There are several reasons that we could reach here:
          //1.StoreFindToken is Uninitialized, and we have more index segments than journal can cover.
          //2.StoreFindToken is JournalBased, but the journal somehow moved too much ahead, the offset in the FindToken is no longer within journal.
          //  2.1 There might just be lots of new entries coming to journal so the journal moves ahead.
          //  2.2 The journal might be cleared and new log segment is created for compaction (auto-close last segment)
          if (entry != null && !entry.getKey().equals(indexSegments.lastKey()) || isJournalEmptyAfterAutoClose()) {
            startTimeInMs = time.milliseconds();
            newToken = findEntriesFromSegmentStartOffset(entry.getKey(), null, messageEntries,
                new FindEntriesCondition(maxTotalSizeOfEntries), indexSegments, storeToken.getInclusive());
            logger.trace("Journal based to segment based token, Time used to find entries: {}",
                (time.milliseconds() - startTimeInMs));

            startTimeInMs = time.milliseconds();
            updateStateForMessages(messageEntries);
            logger.trace("Journal based to segment based token, Time used to update state: {}",
                (time.milliseconds() - startTimeInMs));
          } else {
            newToken = storeToken;
          }
          startTimeInMs = time.milliseconds();
          eliminateDuplicates(messageEntries);
          logger.trace("Journal based to segment based token, Time used to eliminate duplicates: {}",
              (time.milliseconds() - startTimeInMs));
          logger.trace("Index [{}]: new FindInfo [{}]", dataDir, newToken);
          long totalBytesRead = getTotalBytesRead(newToken, messageEntries, logEndOffsetBeforeFind, indexSegments);
          newToken.setBytesRead(totalBytesRead);
          return new FindInfo(messageEntries, newToken);
        }
      } else {
        // Index based token
        // Find the index segment corresponding to the token indexStartOffset.
        // Get entries starting from the token Key in this index.
        startTimeInMs = time.milliseconds();
        StoreFindToken newToken;
        //if index based token points to the last entry of last index segment, and journal size cleaned up after last log
        //segment auto close, we should return the original index based token instead of getting from journal.
        if (isJournalEmptyAfterAutoClose() && indexBasedTokenPointsToLastEntryOfLastIndexSegment(storeToken,
            indexSegments)) {
          newToken = storeToken;
        } else {
          newToken = findEntriesFromSegmentStartOffset(storeToken.getOffset(), storeToken.getStoreKey(), messageEntries,
              new FindEntriesCondition(maxTotalSizeOfEntries), indexSegments, storeToken.getInclusive());
        }
        logger.trace("Segment based token, Time used to find entries: {}", (time.milliseconds() - startTimeInMs));

        startTimeInMs = time.milliseconds();
        updateStateForMessages(messageEntries);
        logger.trace("Segment based token, Time used to update state: {}", (time.milliseconds() - startTimeInMs));

        startTimeInMs = time.milliseconds();
        eliminateDuplicates(messageEntries);
        logger.trace("Segment based token, Time used to eliminate duplicates: {}",
            (time.milliseconds() - startTimeInMs));

        long totalBytesRead = getTotalBytesRead(newToken, messageEntries, logEndOffsetBeforeFind, indexSegments);
        newToken.setBytesRead(totalBytesRead);
        return new FindInfo(messageEntries, newToken);
      }
    } catch (StoreException e) {
      throw e;
    } catch (Exception e) {
      throw new StoreException("Unknown error when finding entries for index " + dataDir, e,
          StoreErrorCodes.UnknownError);
    }
  }
Finds all the entries from the given start token(inclusive). The token defines the start position in the index from
where entries needs to be fetched
@param token The token that signifies the start position in the index from where entries need to be retrieved
@param maxTotalSizeOfEntries The maximum total size of entries that needs to be returned. The api will try to
                             return a list of entries whose total size is close to this value.
@return The FindInfo state that contains both the list of entries and the new findtoken to start the next iteration
ambry-store/src/main/java/com/github/ambry/store/PersistentIndex.java