kinesis-lib/src/main/java/com/streamsets/pipeline/stage/origin/kinesis/StreamSetsRecordProcessor.java
@Override
  public void shutdown(ShutdownInput shutdownInput) {
    LOG.info("Shutting down record processor for shard: {}", shardId);

    if (ShutdownReason.TERMINATE.equals(shutdownInput.getShutdownReason())) {
      // Shard is closed / finished processing. Checkpoint all processing up to here.
      try {
        shutdownInput.getCheckpointer().checkpoint();
        LOG.debug("Checkpointed due to record processor shutdown request.");
      } catch (InvalidStateException | ShutdownException e) {
        LOG.error("Error checkpointing batch: {}", e.toString(), e);
      }
    }
  }
@Override
  public void shutdown(ShutdownInput shutdownInput) {
    LOG.info("Shutting down record processor for shard: {}", shardId);

    if (ShutdownReason.TERMINATE.equals(shutdownInput.getShutdownReason())) {
      // Shard is closed / finished processing. Checkpoint all processing up to here.
      try {
        shutdownInput.getCheckpointer().checkpoint();
        LOG.debug("Checkpointed due to record processor shutdown request.");
      } catch (InvalidStateException | ShutdownException e) {
        LOG.error("Error checkpointing batch: {}", e.toString(), e);
      }
    }
  }
We don't checkpoint on SHUTDOWN_REQUESTED because we currently always
checkpoint each batch in {@link #processRecords}.

@param shutdownInput {@inheritDoc}