stagesupport/src/main/java/com/streamsets/pipeline/support/service/ServicesUtil.java
public static List<Record> parseAll(
    Stage.Context stageContext,
    ToErrorContext toErrorContext,
    boolean produceSingleRecordPerMessage,
    String messageId,
    byte[] payload
  ) throws StageException {
    List<Record> records = new ArrayList<>();
    try (DataParser parser = stageContext.getService(DataFormatParserService.class).getParser(messageId, payload)) {
      Record record = null;
      do {
        try {
          record = parser.parse();
        } catch (RecoverableDataParserException e) {
          handleException(stageContext, toErrorContext, messageId, e, e.getUnparsedRecord());
          //Go to next record
          continue;
        }
        if (record != null) {
          records.add(record);
        }
      } while (record != null);
    } catch (IOException |DataParserException ex) {
      Record record = stageContext.createRecord(messageId);
      record.set(Field.create(payload));
      handleException(stageContext, toErrorContext, messageId, ex, record);
      return records;
    }
    if (produceSingleRecordPerMessage) {
      List<Field> list = new ArrayList<>();
      for (Record record : records) {
        list.add(record.get());
      }
      Record record = records.get(0);
      record.set(Field.create(list));
      records.clear();
      records.add(record);
    }
    return records;
  }
public static List<Record> parseAll(
    Stage.Context stageContext,
    ToErrorContext toErrorContext,
    boolean produceSingleRecordPerMessage,
    String messageId,
    byte[] payload
  ) throws StageException {
    List<Record> records = new ArrayList<>();
    try (DataParser parser = stageContext.getService(DataFormatParserService.class).getParser(messageId, payload)) {
      Record record = null;
      do {
        try {
          record = parser.parse();
        } catch (RecoverableDataParserException e) {
          handleException(stageContext, toErrorContext, messageId, e, e.getUnparsedRecord());
          //Go to next record
          continue;
        }
        if (record != null) {
          records.add(record);
        }
      } while (record != null);
    } catch (IOException |DataParserException ex) {
      Record record = stageContext.createRecord(messageId);
      record.set(Field.create(payload));
      handleException(stageContext, toErrorContext, messageId, ex, record);
      return records;
    }
    if (produceSingleRecordPerMessage) {
      List<Field> list = new ArrayList<>();
      for (Record record : records) {
        list.add(record.get());
      }
      Record record = records.get(0);
      record.set(Field.create(list));
      records.clear();
      records.add(record);
    }
    return records;
  }
public static List < Record > parseAll ( Stage . Context stageContext , ToErrorContext toErrorContext , boolean produceSingleRecordPerMessage , String messageId , byte [ ] payload ) throws StageException { List < Record > records = new ArrayList <> ( ) ; try ( DataParser parser = stageContext . getService ( DataFormatParserService . class ) . getParser ( messageId , payload ) ) { Record record = null ; do { try { record = parser . parse ( ) ; } catch ( RecoverableDataParserException e ) { handleException ( stageContext , toErrorContext , messageId , e , e . getUnparsedRecord ( ) ) ; //Go to next record continue ; } if ( record != null ) { records . add ( record ) ; } } while ( record != null ) ; } catch ( IOException | DataParserException ex ) { Record record = stageContext . createRecord ( messageId ) ; record . set ( Field . create ( payload ) ) ; handleException ( stageContext , toErrorContext , messageId , ex , record ) ; return records ; } if ( produceSingleRecordPerMessage ) { List < Field > list = new ArrayList <> ( ) ; for ( Record record : records ) { list . add ( record . get ( ) ) ; } Record record = records . get ( 0 ) ; record . set ( Field . create ( list ) ) ; records . clear ( ) ; records . add ( record ) ; } return records ; }
Convenience method to parse all available records in given payload. This method assumes that the stage already
depends on DataFormatParserService service and loads the service instance from Stage.Context.

@param stageContext Stage's context
@param toErrorContext Error context for any parser errors
@param produceSingleRecordPerMessage If set, then output will be list of exactly one record
@param messageId Message id base to be used with generated records
@param payload Binary payload to parse
@return
@throws StageException
Convenience method to parse all available records in given payload . This method assumes that the stage already depends on DataFormatParserService service and loads the service instance from Stage . Context .