private void testShutdownDuringReshardHelper(int numRecords)
      throws InterruptedException {
    String system = "kinesis";
    String stream = "stream";
    final CountDownLatch receivedShutdownLatch = new CountDownLatch(1);
    final CountDownLatch receivedRecordsLatch = new CountDownLatch(numRecords > 0 ? 1 : 0);

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

      @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 processRecords on the processor
    List<Record> records = generateRecords(numRecords, Collections.singletonList(processor)).get(processor);

    // Verification steps

    // Verify there is a receivedRecords call to listener.
    Assert.assertEquals("Unable to receive records.", 0, receivedRecordsLatch.getCount());

    // Call shutdown (with TERMINATE reason) on processor and verify that the processor does not call shutdown on the
    // listener until checkpoint is called for the last record consumed from shard.
    new Thread(() -> shutDownProcessor(processor, ShutdownReason.TERMINATE)).start();

    // If there are no records, the processor should shutdown immediately.
    if (numRecords == 0) {
      Assert.assertTrue("Unable to shutdown processor.",
          receivedShutdownLatch.await(MAX_WAIT_TIME_SHUTDOWN_RECEIVED_MS, TimeUnit.MILLISECONDS));
      return;
    }

    Assert.assertFalse("Processor shutdown too early.",
        receivedShutdownLatch.await(MAX_WAIT_TIME_SHUTDOWN_RECEIVED_MS, TimeUnit.MILLISECONDS));

    // Call checkpoint for the last but one record and the processor should still not call shutdown on listener.
    processor.checkpoint(records.get(records.size() - 2).getSequenceNumber());
    Assert.assertFalse("Processor shutdown too early.",
        receivedShutdownLatch.await(MAX_WAIT_TIME_SHUTDOWN_RECEIVED_MS, TimeUnit.MILLISECONDS));

    // Call checkpoint for the last record and the parent partition should be removed from mapper.
    processor.checkpoint(records.get(records.size() - 1).getSequenceNumber());
    Assert.assertTrue("Unable to shutdown processor.",
        receivedShutdownLatch.await(MAX_WAIT_TIME_SHUTDOWN_RECEIVED_MS, TimeUnit.MILLISECONDS));
  }
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