Map<SystemStreamPartition, List<IncomingMessageEnvelope>> poll(Map<SystemStreamPartition, String> sspsToOffsets) {
    return sspsToOffsets.entrySet()
        .stream()
        .collect(Collectors.toMap(Map.Entry::getKey, entry -> poll(entry.getKey(), entry.getValue())));
  }
Handles the poll request from {@link InMemorySystemConsumer}. It uses the input offset as the starting offset for
each {@link SystemStreamPartition}.

@param sspsToOffsets ssps to offset mapping

@return a {@link Map} of {@link SystemStreamPartition} to {@link List} of {@link IncomingMessageEnvelope}
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java