private Map<String, KinesisRecordProcessor> createAndInitProcessors(IRecordProcessorFactory factory, int numShards) {
    Map<String, KinesisRecordProcessor> processorMap = new HashMap<>();
    IntStream.range(0, numShards)
        .forEach(p -> {
          String shardId = String.format("shard-%05d", p);
          // Create Kinesis processor
          KinesisRecordProcessor processor = (KinesisRecordProcessor) factory.createProcessor();

          // Initialize the shard
          ExtendedSequenceNumber seqNum = new ExtendedSequenceNumber("0000");
          InitializationInput initializationInput =
              new InitializationInput().withShardId(shardId).withExtendedSequenceNumber(seqNum);
          processor.initialize(initializationInput);
          processorMap.put(shardId, processor);
        });
    return processorMap;
  }
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