@Override
  public IncomingMessageEnvelope next() {
    refresh();

    if (peeks.size() == 0 || (endOffset != null && admin.offsetComparator(peeks.peek().getOffset(), endOffset) > 0)) {
      throw new NoSuchElementException();
    }

    return peeks.poll();
  }
Iterates over messages in the provided {@link SystemStreamPartition} using the provided
{@link SystemConsumer} until all messages with offsets up to and including the {@code endOffset} have been consumed.
If {@code endOffset} is null, the iterator will return all messages until caught up to head.
samza-api/src/main/java/org/apache/samza/system/BoundedSSPIterator.java