public static <T> CsvIOParse<T> parse(Class<T> klass, CSVFormat csvFormat) {
    CsvIOParseHelpers.validateCsvFormat(csvFormat);
    SchemaProvider provider = new DefaultSchema.DefaultSchemaProvider();
    TypeDescriptor<T> type = TypeDescriptor.of(klass);
    Schema schema =
        checkStateNotNull(
            provider.schemaFor(type),
            "Illegal %s: Schema could not be generated from given %s class",
            Schema.class,
            klass);
    CsvIOParseHelpers.validateCsvFormatWithSchema(csvFormat, schema);
    SerializableFunction<Row, T> fromRowFn =
        checkStateNotNull(
            provider.fromRowFunction(type),
            "FromRowFn could not be generated from the given %s class",
            klass);
    SerializableFunction<T, Row> toRowFn =
        checkStateNotNull(
            provider.toRowFunction(type),
            "ToRowFn could not be generated from the given %s class",
            klass);
    SchemaCoder<T> coder = SchemaCoder.of(schema, type, toRowFn, fromRowFn);
    CsvIOParseConfiguration.Builder<T> builder = CsvIOParseConfiguration.builder();
    builder.setCsvFormat(csvFormat).setSchema(schema).setCoder(coder).setFromRowFn(fromRowFn);
    return CsvIOParse.<T>builder().setConfigBuilder(builder).build();
  }
@Test
  public void parsesPOJOs() {
    Pipeline pipeline = Pipeline.create();
    PCollection<String> input =
        csvRecords(
            pipeline,
            "# This is a comment",
            "aBoolean,aDouble,aFloat,anInteger,aLong,aString",
            "true,1.0,2.0,3,4,foo",
            "N/A,6.0,7.0,8,9,bar",
            "false,12.0,14.0,8,24,\"foo\nbar\"",
            "true,1.0,2.0,3,4,foo$,bar");
    List<SchemaAwareJavaBeans.NullableAllPrimitiveDataTypes> want =
        Arrays.asList(
            nullableAllPrimitiveDataTypes(true, 1.0d, 2.0f, 3, 4L, "foo"),
            nullableAllPrimitiveDataTypes(null, 6.0d, 7.0f, 8, 9L, "bar"),
            nullableAllPrimitiveDataTypes(false, 12.0d, 14.0f, 8, 24L, "foo\nbar"),
            nullableAllPrimitiveDataTypes(true, 1.0d, 2.0f, 3, 4L, "foo,bar"));

    CsvIOParse<NullableAllPrimitiveDataTypes> underTest =
        CsvIO.parse(NullableAllPrimitiveDataTypes.class, csvFormat());
    CsvIOParseResult<NullableAllPrimitiveDataTypes> result = input.apply(underTest);
    PAssert.that(result.getOutput()).containsInAnyOrder(want);
    PAssert.that(result.getErrors()).empty();

    pipeline.run();
  }