public static CsvIOParse<Row> parseRows(Schema schema, CSVFormat csvFormat) {
    CsvIOParseHelpers.validateCsvFormat(csvFormat);
    CsvIOParseHelpers.validateCsvFormatWithSchema(csvFormat, schema);
    RowCoder coder = RowCoder.of(schema);
    CsvIOParseConfiguration.Builder<Row> builder = CsvIOParseConfiguration.builder();
    builder.setCsvFormat(csvFormat).setSchema(schema).setCoder(coder).setFromRowFn(row -> row);
    return CsvIOParse.<Row>builder().setConfigBuilder(builder).build();
  }
@Test
  public void givenStringToRecordError_RecordToObjectError_emits() {
    Pipeline pipeline = Pipeline.create();
    PCollection<String> input =
        pipeline.apply(
            Create.of("true,\"1.1,3.141592,1,5,foo", "true,1.1,3.141592,this_is_an_error,5,foo"));
    Schema schema =
        Schema.builder()
            .addBooleanField("aBoolean")
            .addDoubleField("aDouble")
            .addFloatField("aFloat")
            .addInt32Field("anInteger")
            .addInt64Field("aLong")
            .addStringField("aString")
            .build();
    CsvIOParse<Row> underTest = CsvIO.parseRows(schema, csvFormat().withQuote('"'));
    CsvIOParseResult<Row> result = input.apply(underTest);
    PAssert.thatSingleton(result.getErrors().apply(Count.globally())).isEqualTo(2L);
    PAssert.thatSingleton(
            stackTraceContains(result.getErrors(), CsvIOStringToCsvRecord.class.getName()))
        .isEqualTo(1L);
    PAssert.thatSingleton(
            stackTraceContains(result.getErrors(), CsvIORecordToObjects.class.getName()))
        .isEqualTo(1L);

    pipeline.run();
  }