kinesis-lib/src/main/java/com/streamsets/pipeline/stage/origin/kinesis/StreamSetsRecordProcessor.java
@Override
  public void processRecords(ProcessRecordsInput processRecordsInput) {
    LOG.debug("RecordProcessor processRecords called");

    try {
      IRecordProcessorCheckpointer checkpointer = processRecordsInput.getCheckpointer();
      startBatch();
      Optional<Record> lastProcessedRecord = Optional.empty();
      int recordCount = 0;
      for (Record kRecord : processRecordsInput.getRecords()) {
        try {
          KinesisUtil.processKinesisRecord(shardId, kRecord, parserFactory).forEach(batchMaker::addRecord);
          lastProcessedRecord = Optional.of(kRecord);
          if (++recordCount == maxBatchSize) {
            recordCount = 0;
            finishBatch(checkpointer, kRecord);
            startBatch();
          }
        } catch (DataParserException | IOException e) {
          com.streamsets.pipeline.api.Record record = context.createRecord(kRecord.getSequenceNumber());
          record.set(Field.create(kRecord.getData().array()));
          try {
            errorRecordHandler.onError(new OnRecordErrorException(
                record,
                Errors.KINESIS_03,
                kRecord.getSequenceNumber(),
                e.toString(),
                e
            ));
            // move the lastProcessedRecord forward if not set to stop pipeline
            lastProcessedRecord = Optional.of(kRecord);
          } catch (StageException ex) {
            // KCL skips over the data records that were passed prior to the exception
            // that is, these records are not re-sent to this record processor
            // or to any other record processor in the consumer.
            lastProcessedRecord.ifPresent(r -> finishBatch(checkpointer, r));
            try {
              error.put(ex);
            } catch (InterruptedException ie) {
              Thread.currentThread().interrupt();
            }
            return;
          }
        }
      }
      lastProcessedRecord.ifPresent(r -> finishBatch(checkpointer, r));
    } catch (Exception e) {
      LOG.error("Unknown error while processing records: {}", e.toString(), e);
      context.reportError(Errors.KINESIS_17, e.toString());
    }
  }
@Override
  public void processRecords(ProcessRecordsInput processRecordsInput) {
    LOG.debug("RecordProcessor processRecords called");

    try {
      IRecordProcessorCheckpointer checkpointer = processRecordsInput.getCheckpointer();
      startBatch();
      Optional<Record> lastProcessedRecord = Optional.empty();
      int recordCount = 0;
      for (Record kRecord : processRecordsInput.getRecords()) {
        try {
          KinesisUtil.processKinesisRecord(shardId, kRecord, parserFactory).forEach(batchMaker::addRecord);
          lastProcessedRecord = Optional.of(kRecord);
          if (++recordCount == maxBatchSize) {
            recordCount = 0;
            finishBatch(checkpointer, kRecord);
            startBatch();
          }
        } catch (DataParserException | IOException e) {
          com.streamsets.pipeline.api.Record record = context.createRecord(kRecord.getSequenceNumber());
          record.set(Field.create(kRecord.getData().array()));
          try {
            errorRecordHandler.onError(new OnRecordErrorException(
                record,
                Errors.KINESIS_03,
                kRecord.getSequenceNumber(),
                e.toString(),
                e
            ));
            // move the lastProcessedRecord forward if not set to stop pipeline
            lastProcessedRecord = Optional.of(kRecord);
          } catch (StageException ex) {
            // KCL skips over the data records that were passed prior to the exception
            // that is, these records are not re-sent to this record processor
            // or to any other record processor in the consumer.
            lastProcessedRecord.ifPresent(r -> finishBatch(checkpointer, r));
            try {
              error.put(ex);
            } catch (InterruptedException ie) {
              Thread.currentThread().interrupt();
            }
            return;
          }
        }
      }
      lastProcessedRecord.ifPresent(r -> finishBatch(checkpointer, r));
    } catch (Exception e) {
      LOG.error("Unknown error while processing records: {}", e.toString(), e);
      context.reportError(Errors.KINESIS_17, e.toString());
    }
  }
{@inheritDoc}