@Override
  public void processRecords(ProcessRecordsInput processRecordsInput) {
    // KCL does not send any records to the processor that was shutdown.
    Validate.isTrue(!shutdownRequested,
        String.format("KCL returned records after shutdown is called on the processor %s.", this));
    // KCL aways gives reference to the same checkpointer instance for a given processor instance.
    checkpointer = processRecordsInput.getCheckpointer();
    List<Record> records = processRecordsInput.getRecords();
    // Empty records are expected when KCL config has CallProcessRecordsEvenForEmptyRecordList set to true.
    if (!records.isEmpty()) {
      lastProcessedRecordSeqNumber = new ExtendedSequenceNumber(records.get(records.size() - 1).getSequenceNumber());
      listener.onReceiveRecords(ssp, records, processRecordsInput.getMillisBehindLatest());
    }
  }
Process data records. The Amazon Kinesis Client Library will invoke this method to deliver data records to the
application. Upon fail over, the new instance will get records with sequence number greater than the checkpoint
position for each partition key.

@param processRecordsInput Provides the records to be processed as well as information and capabilities related
       to them (eg checkpointing).
samza-aws/src/main/java/org/apache/samza/system/kinesis/consumer/KinesisRecordProcessor.java