private List<IncomingMessageEnvelope> poll(SystemStreamPartition ssp, String offset) {
    int startingOffset = Integer.parseInt(offset);
    List<IncomingMessageEnvelope> messageEnvelopesForSSP = bufferedMessages.getOrDefault(ssp, new LinkedList<>());

    if (startingOffset >= messageEnvelopesForSSP.size()) {
      return new ArrayList<>();
    }

    return ImmutableList.copyOf(messageEnvelopesForSSP.subList(startingOffset, messageEnvelopesForSSP.size()));
  }
Fetch partition count for the given input stream stream.

@param systemStream input system stream

@return the partition count if available or {@link InMemoryManager#DEFAULT_PARTITION_COUNT}
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java