@Test
  public void testCheckpointAfterInit() {
    String system = "kinesis";
    String stream = "stream";
    final CountDownLatch receivedShutdownLatch = new CountDownLatch(1);

    KinesisRecordProcessorListener listener = new KinesisRecordProcessorListener() {
      @Override
      public void onReceiveRecords(SystemStreamPartition ssp, List<Record> records, long millisBehindLatest) {
      }

      @Override
      public void onShutdown(SystemStreamPartition ssp) {
        receivedShutdownLatch.countDown();
      }
    };

    KinesisRecordProcessor processor =
        new KinesisRecordProcessor(new SystemStreamPartition(system, stream, new Partition(0)), listener);

    // Initialize the processor
    ExtendedSequenceNumber seqNum = new ExtendedSequenceNumber("0000");
    InitializationInput initializationInput =
        new InitializationInput().withShardId("shard-0000").withExtendedSequenceNumber(seqNum);
    processor.initialize(initializationInput);

    // Call checkpoint. This checkpoint could have originally headed to the processor instance for the same shard but
    // due to reassignment a new processor instance is created.
    processor.checkpoint("1234567");


    // Call shutdown (with ZOMBIE reason) on processor and verify that the processor calls shutdown on the listener.
    shutDownProcessor(processor, ShutdownReason.ZOMBIE);

    // Verify that the processor is shutdown.
    Assert.assertEquals("Unable to shutdown processor.", 0, receivedShutdownLatch.getCount());
  }
Test the scenario where a processor instance is created for a shard and while it is processing records, it got
re-assigned to the same consumer. This results in a new processor instance owning the shard and this instance
could receive checkpoint calls for the records that are processed by the old processor instance. This test covers
the scenario where the new instance receives the checkpoint call while it is done with the initialization phase and
before it processed any records.
samza-aws/src/test/java/org/apache/samza/system/kinesis/consumer/TestKinesisRecordProcessor.java