@Override
  public Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(
      Set<SystemStreamPartition> systemStreamPartitions, long timeout) throws InterruptedException {
    Map<SystemStreamPartition, String> sspOffsetPairToFetch = sspToOffset.entrySet()
        .stream()
        .filter(entry -> systemStreamPartitions.contains(entry.getKey()))
        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

    Map<SystemStreamPartition, List<IncomingMessageEnvelope>> result = memoryManager.poll(sspOffsetPairToFetch);

    for (Map.Entry<SystemStreamPartition, List<IncomingMessageEnvelope>> sspToMessage : result.entrySet()) {
      sspToOffset.computeIfPresent(sspToMessage.getKey(), (ssp, offset) -> {
        int newOffset = Integer.parseInt(offset) + sspToMessage.getValue().size();
        return String.valueOf(newOffset);
      });
      // absent should never be the case
    }

    return result;
  }
Poll the SystemConsumer to get any available messages from the underlying
system.

<p>
If the underlying implementation does not take care to adhere to the
timeout parameter, the SamzaContainer's performance will suffer
drastically. Specifically, if poll blocks when it's not supposed to, it
will block the entire main thread in SamzaContainer, and no messages will
be processed while blocking is occurring.
</p>

@param systemStreamPartitions
         A set of SystemStreamPartition to poll for new messages. If
         SystemConsumer has messages available for other registered
         SystemStreamPartitions, but they are not in the
         systemStreamPartitions set in a given poll invocation, they can't
         be returned. It is illegal to pass in SystemStreamPartitions that
         have not been registered with the SystemConsumer first.
@param timeout
         If timeout &lt; 0, poll will block unless all SystemStreamPartition
         are at "head" (the underlying system has been checked, and
         returned an empty set). If at head, an empty map is returned. If
         timeout &gt;= 0, poll will return any messages that are currently
         available for any of the SystemStreamPartitions specified. If no
         new messages are available, it will wait up to timeout
         milliseconds for messages from any SystemStreamPartition to become
         available. It will return an empty map if the timeout is hit, and
         no new messages are available.
@return A map from SystemStreamPartitions to any available
        IncomingMessageEnvelopes for the SystemStreamPartitions. If no
        messages are available for a SystemStreamPartition that was
        supplied in the polling set, the map will not contain a key for the
        SystemStreamPartition. Will return an empty map, not null, if no
        new messages are available for any SystemStreamPartitions in the
        input set.
@throws InterruptedException
         Thrown when a blocking poll has been interrupted by another
         thread.
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemorySystemConsumer.java