container/src/main/java/com/streamsets/datacollector/runner/production/BadRecordsHandler.java
private List<Record> getBadRecords(ErrorSink errorSink) {
    List<Record> badRecords = new ArrayList<>();

    for (Map.Entry<String, List<Record>> entry : errorSink.getErrorRecords().entrySet()) {
      for (Record record : entry.getValue()) {
        RecordImpl errorRecord;

        switch (errorRecordPolicy) {
          case ORIGINAL_RECORD:
            errorRecord = (RecordImpl) ((RecordImpl)record).getHeader().getSourceRecord();
            errorRecord.getHeader().copyErrorFrom(record);
            break;
          case STAGE_RECORD:
            errorRecord = (RecordImpl) record;
            break;
          default:
           throw new IllegalArgumentException("Uknown error record policy: " + errorRecordPolicy);
        }

        errorRecord.getHeader().setErrorContext(runtimeInfo.getId(), pipelineName);
        badRecords.add(errorRecord);
      }
    }
    return badRecords;
  }
private List<Record> getBadRecords(ErrorSink errorSink) {
    List<Record> badRecords = new ArrayList<>();

    for (Map.Entry<String, List<Record>> entry : errorSink.getErrorRecords().entrySet()) {
      for (Record record : entry.getValue()) {
        RecordImpl errorRecord;

        switch (errorRecordPolicy) {
          case ORIGINAL_RECORD:
            errorRecord = (RecordImpl) ((RecordImpl)record).getHeader().getSourceRecord();
            errorRecord.getHeader().copyErrorFrom(record);
            break;
          case STAGE_RECORD:
            errorRecord = (RecordImpl) record;
            break;
          default:
           throw new IllegalArgumentException("Uknown error record policy: " + errorRecordPolicy);
        }

        errorRecord.getHeader().setErrorContext(runtimeInfo.getId(), pipelineName);
        badRecords.add(errorRecord);
      }
    }
    return badRecords;
  }
private List < Record > getBadRecords ( ErrorSink errorSink ) { List < Record > badRecords = new ArrayList <> ( ) ; for ( Map . Entry < String , List < Record > > entry : errorSink . getErrorRecords ( ) . entrySet ( ) ) { for ( Record record : entry . getValue ( ) ) { RecordImpl errorRecord ; switch ( errorRecordPolicy ) { case ORIGINAL_RECORD : errorRecord = ( RecordImpl ) ( ( RecordImpl ) record ) . getHeader ( ) . getSourceRecord ( ) ; errorRecord . getHeader ( ) . copyErrorFrom ( record ) ; break ; case STAGE_RECORD : errorRecord = ( RecordImpl ) record ; break ; default : throw new IllegalArgumentException ( "Uknown error record policy: " + errorRecordPolicy ) ; } errorRecord . getHeader ( ) . setErrorContext ( runtimeInfo . getId ( ) , pipelineName ) ; badRecords . add ( errorRecord ) ; } } return badRecords ; }
Generate list of error records from the error sink. What precise records will be returned depends on the error
record policy configuration.

@param errorSink Error sink with the error records
@return List with records that should be sent to error stream
Generate list of error records from the error sink . What precise records will be returned depends on the error record policy configuration .