private StoreFindToken findEntriesFromSegmentStartOffset(Offset initialSegmentStartOffset, StoreKey key,
      List<MessageInfo> messageEntries, FindEntriesCondition findEntriesCondition,
      ConcurrentSkipListMap<Offset, IndexSegment> indexSegments, boolean inclusive) throws StoreException {
    Offset segmentStartOffset = initialSegmentStartOffset;
    //Always checking journal before indexSegments to avoid race condition since when we put new entry, the index segment
    //will be updated first.
    if (!isJournalEmptyAfterAutoClose() && segmentStartOffset.equals(indexSegments.lastKey())) {
      //We would never have given away a token with a segmentStartOffset of the latest segment unless the journal has
      //been cleaned up by (auto close last log segment).
      throw new IllegalArgumentException(
          "Index : " + dataDir + " findEntriesFromOffset segment start offset " + segmentStartOffset
              + " is of the last segment");
    }

    Offset newTokenSegmentStartOffset = null;
    Offset newTokenOffsetInJournal = null;

    IndexSegment segmentToProcess = indexSegments.get(segmentStartOffset);
    AtomicLong currentTotalSizeOfEntries = new AtomicLong(0);

    /* First, get keys from the segment corresponding to the passed in segment start offset if the token has a non-null
       key. Otherwise, since all the keys starting from the offset have to be read, skip this and check in the journal
       first. */
    if (key != null) {
      if (segmentToProcess.getEntriesSince(key, findEntriesCondition, messageEntries, currentTotalSizeOfEntries,
          inclusive)) {
        // if we did fetch entries from this segment, set the new token info accordingly.
        newTokenSegmentStartOffset = segmentStartOffset;
      }
      logger.trace("Index : {} findEntriesFromOffset segment start offset {} with key {} total entries received {}",
          dataDir, segmentStartOffset, key, messageEntries.size());
      // Notice that we might not finish scanning the segmentToProcess, it's possible that current index segment contains
      // enough index values so that the findEntriesCondition returns false. But we are still moving to the next index segment
      // because when findEntriesCondition returns false, we will skip the while loop below and return the new find token
      // right away. And if findEntriesCondition returns true, that means we finish scanning of the current index segment so
      // we can move to the next one. If key is not null and the segmentStartOffset equals to the last index segment's start offset,
      // segmentToProcess will be set to null since the index based token has pointed to the last index segment and at this moment journal
      // should be null so we will skip the while loop below and return the new index based token right away.
      segmentStartOffset = indexSegments.higherKey(segmentStartOffset);
      segmentToProcess = segmentStartOffset == null ? null : indexSegments.get(segmentStartOffset);
    }
    while (segmentToProcess != null && findEntriesCondition.proceed(currentTotalSizeOfEntries.get(),
        segmentToProcess.getLastModifiedTimeSecs())) {
      // Check in the journal to see if we are already at an offset in the journal, if so get entries from it.
      Offset journalFirstOffsetBeforeCheck = journal.getFirstOffset();
      Offset journalLastOffsetBeforeCheck = journal.getLastOffset();
      // If segmentStartOffset is null, then the journal has to be empty and in journal.getEntriesSince method the
      // first entry in journal will be null and null will be returned right away.
      List<JournalEntry> entries = journal.getEntriesSince(segmentStartOffset, true);
      Offset endOffsetOfSnapshot = getCurrentEndOffset(indexSegments);
      if (entries != null) {
        logger.trace("Index : {} findEntriesFromOffset journal offset {} total entries received {}", dataDir,
            segmentStartOffset, entries.size());
        IndexSegment currentSegment = segmentToProcess;
        Map<StoreKey, MessageInfo> messageInfoCache = new HashMap<>();
        for (JournalEntry entry : entries) {
          if (entry.getOffset().compareTo(currentSegment.getEndOffset()) > 0) {
            Offset nextSegmentStartOffset = indexSegments.higherKey(currentSegment.getStartOffset());
            // since we were given a snapshot of indexSegments, it is not guaranteed that the snapshot contains all of
            // the entries obtained from journal.getEntriesSince() that was performed after the snapshot was obtained.
            // Therefore, it is possible that some of the entries obtained from the journal don't exist in the snapshot
            // Such entries can be detected when a rollover is required and we have no more index segments left to roll
            // over to.
            if (nextSegmentStartOffset == null) {
              // stop if there are no more index segments in this ref.
              break;
            }
            currentSegment = indexSegments.get(nextSegmentStartOffset);
            // stop if ineligible because of last modified time
            if (!findEntriesCondition.proceed(currentTotalSizeOfEntries.get(),
                currentSegment.getLastModifiedTimeSecs())) {
              break;
            }
          }
          newTokenOffsetInJournal = entry.getOffset();
          addMessageInfoForJournalEntry(entry, endOffsetOfSnapshot, indexSegments, messageInfoCache, messageEntries,
              currentTotalSizeOfEntries);
          if (!findEntriesCondition.proceed(currentTotalSizeOfEntries.get(),
              currentSegment.getLastModifiedTimeSecs())) {
            break;
          }
        }
        break; // we have entered and finished reading from the journal, so we are done.
      }

      //If journal is empty due to auto close last log segment, there's no need to check this.
      //Always checking journal before indexSegments to avoid race condition since when we put new entry, the index segment
      //will be updated first.
      if (!isJournalEmptyAfterAutoClose() && segmentStartOffset.equals(validIndexSegments.lastKey())) {
        /* The start offset is of the latest segment, and was not found in the journal. This means an entry was added
         * to the index (creating a new segment) but not yet to the journal. However, if the journal does not contain
         * the latest segment's start offset, then it *must* have the previous segment's start offset (if it did not,
         * then the previous segment must have been the latest segment at the time it was scanned, which couldn't have
         * happened due to this same argument).
         * */
        throw new IllegalStateException(
            "Index : " + dataDir + " findEntriesFromOffset segment start offset " + segmentStartOffset
                + " is of the latest segment and not found in journal with range [" + journalFirstOffsetBeforeCheck
                + ", " + journalLastOffsetBeforeCheck + "]");
      } else {
        // Read and populate from the first key in the segment with this segmentStartOffset
        if (segmentToProcess.getEntriesSince(null, findEntriesCondition, messageEntries, currentTotalSizeOfEntries,
            false)) {
          newTokenSegmentStartOffset = segmentStartOffset;
        }
        logger.trace(
            "Index : {} findEntriesFromOffset segment start offset {} with all the keys, total entries received {}",
            dataDir, segmentStartOffset, messageEntries.size());
        segmentStartOffset = indexSegments.higherKey(segmentStartOffset);
        if (segmentStartOffset == null) {
          // we have reached the last index segment in the index ref we have. The journal has moved past this final
          // segment and we have nothing to do except return what we have. Our view is still consistent and "up-to-date"
          // w.r.t the time the call was made.
          break;
        }
        segmentToProcess = indexSegments.get(segmentStartOffset);
      }
    }
    if (newTokenOffsetInJournal != null) {
      IndexSegment segmentOfToken = indexSegments.floorEntry(newTokenOffsetInJournal).getValue();
      return new StoreFindToken(newTokenOffsetInJournal, sessionId, incarnationId, false, segmentOfToken.getResetKey(),
          segmentOfToken.getResetKeyType(), segmentOfToken.getResetKeyLifeVersion());
    } else if (messageEntries.size() == 0 && !findEntriesCondition.hasEndTime()) {
      // If the condition does not have an end time, then since we have entered a segment, we should return at least one
      // message unless the new log segment only contains headers.
      throw new IllegalStateException(
          "Message entries cannot be null. At least one entry should have been returned, start offset: "
              + initialSegmentStartOffset + ", key: " + key + ", findEntriesCondition: " + findEntriesCondition);
    } else {
      if (newTokenSegmentStartOffset == null) {
        return new StoreFindToken();
      }
      // if newTokenSegmentStartOffset is set, then we did fetch entries from that segment, otherwise return an
      // uninitialized token
      IndexSegment segmentOfToken = indexSegments.get(newTokenSegmentStartOffset);
      return new StoreFindToken(messageEntries.get(messageEntries.size() - 1).getStoreKey(), newTokenSegmentStartOffset,
          sessionId, incarnationId, segmentOfToken.getResetKey(), segmentOfToken.getResetKeyType(),
          segmentOfToken.getResetKeyLifeVersion());
    }
  }
Finds entries starting from a key from the segment with start offset initialSegmentStartOffset. The key represents
the position in the segment starting from where entries needs to be fetched.
@param initialSegmentStartOffset The segment start offset of the segment to start reading entries from.
@param key The key representing the position in the segment to start reading entries from (whether to read this key
          as well depends on inclusive variable). If the key is null, all the keys will be read.
@param messageEntries the list to be populated with the MessageInfo for every entry that is read.
@param findEntriesCondition that determines whether to fetch more entries based on a maximum total size of entries
                            that needs to be returned and a time that determines the latest segment to be scanned.
@param indexSegments the map of index segment start {@link Offset} to {@link IndexSegment} instances
@param inclusive whether to start reading entries from the key, {@code true} inclusive, {@code false} exclusive.
@return A token representing the position in the segment/journal up to which entries have been read and returned.
ambry-store/src/main/java/com/github/ambry/store/PersistentIndex.java