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());
    }
  }
@ 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}
{