@Override
  public void shutdown(ShutdownInput shutdownInput) {
    LOG.info("Shutting down {} with reason:{}", this, shutdownInput.getShutdownReason());

    Validate.isTrue(!shutdownRequested, String.format("KCL called shutdown more than once for processor %s.", this));
    shutdownRequested = true;
    // shutdown reason TERMINATE indicates that the shard is closed due to re-sharding. It also indicates that all the
    // records from the shard have been delivered to the consumer and the consumer is expected to checkpoint the
    // progress.
    if (shutdownInput.getShutdownReason() == ShutdownReason.TERMINATE) {
      // We need to ensure that all records are processed and checkpointed before going ahead and marking the processing
      // complete by calling checkpoint() on KCL. We need to checkpoint the completion state for parent shard, for KCL
      // to consume from the child shard(s).
      try {
        LOG.info("Waiting for all the records for {} to be processed.", this);
        // Let's poll periodically and block until the last processed record is checkpointed. Also handle the case
        // where there are no records received for the processor, in which case the lastProcessedRecordSeqNumber will
        // be null.
        while (lastProcessedRecordSeqNumber != null
            && !lastProcessedRecordSeqNumber.equals(lastCheckpointedRecordSeqNumber)) {
          Thread.sleep(POLL_INTERVAL_DURING_PARENT_SHARD_SHUTDOWN_MS);
        }
        LOG.info("Final checkpoint for {} before shutting down.", this);
        shutdownInput.getCheckpointer().checkpoint();
      } catch (Exception e) {
        LOG.warn("An error occurred while committing the final checkpoint in the parent shard {}", this, e);
      }
    }
    listener.onShutdown(ssp);
  }
Invoked by the Amazon Kinesis Client Library to indicate it will no longer send data records to this
RecordProcessor instance.

@param shutdownInput Provides information and capabilities (eg checkpointing) related to shutdown of this record
       processor.
samza-aws/src/main/java/org/apache/samza/system/kinesis/consumer/KinesisRecordProcessor.java