private void setupConsumerMultiplePoll(List<List<IncomingMessageEnvelope>> pollOutputs) throws InterruptedException {
    OngoingStubbing<Map<SystemStreamPartition, List<IncomingMessageEnvelope>>> when =
        when(this.systemConsumer.poll(ImmutableSet.of(CHECKPOINT_SSP), SystemConsumer.BLOCK_ON_OUTSTANDING_MESSAGES));
    for (List<IncomingMessageEnvelope> pollOutput : pollOutputs) {
      when = when.thenReturn(ImmutableMap.of(CHECKPOINT_SSP, pollOutput));
    }
    when.thenReturn(ImmutableMap.of());
  }
Create a new {@link SystemConsumer} that returns a list of messages sequentially at each subsequent poll.

@param pollOutputs a list of poll outputs to be returned at subsequent polls.
                   The i'th call to consumer.poll() will return the list at pollOutputs[i]
samza-kafka/src/test/java/org/apache/samza/checkpoint/kafka/TestKafkaCheckpointManager.java