private void refresh() {
    if (peeks.size() == 0) {
      try {
        Map<SystemStreamPartition, List<IncomingMessageEnvelope>> envelopes = systemConsumer.poll(fetchSet, SystemConsumer.BLOCK_ON_OUTSTANDING_MESSAGES);

        for (List<IncomingMessageEnvelope> systemStreamPartitionEnvelopes : envelopes.values()) {
          peeks.addAll(systemStreamPartitionEnvelopes);
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new SamzaException(e);
      }
    }
  }
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