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;
  }
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