private Map<SystemStreamPartition, List<IncomingMessageEnvelope>> readEvents(Set<SystemStreamPartition> ssps,
      KinesisSystemConsumer consumer, int numEvents) throws InterruptedException {
    Map<SystemStreamPartition, List<IncomingMessageEnvelope>> messages = new HashMap<>();
    int totalEventsConsumed = 0;

    while (totalEventsConsumed < numEvents) {
      Map<SystemStreamPartition, List<IncomingMessageEnvelope>> receivedMessages =
          consumer.poll(ssps, Duration.ofSeconds(1).toMillis());
      receivedMessages.forEach((key, value) -> {
        if (messages.containsKey(key)) {
          messages.get(key).addAll(value);
        } else {
          messages.put(key, new ArrayList<>(value));
        }
      });
      totalEventsConsumed = messages.values().stream().mapToInt(List::size).sum();
    }

    if (totalEventsConsumed < numEvents) {
      String msg = String.format("Received only %d of %d events", totalEventsConsumed, numEvents);
      throw new SamzaException(msg);
    }
    return messages;
  }
Helper to simulate and test the life-cycle of record processing from a kinesis stream with a given number of shards
1. Creation of record processors.
2. Initialization of record processors.
3. Processing records via record processors.
4. Calling checkpoint on record processors.
5. Shutting down (due to re-assignment or lease expiration) record processors.
samza-aws/src/test/java/org/apache/samza/system/kinesis/consumer/TestKinesisSystemConsumer.java