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 givenNonSchemaMappedClass_throws() {
    Pipeline pipeline = Pipeline.create();
    CSVFormat csvFormat =
        CSVFormat.DEFAULT
            .withHeader("a_string", "an_integer", "a_double")
            .withAllowDuplicateHeaderNames(false);
    assertThrows(
        IllegalStateException.class, () -> CsvIO.parse(NonSchemaMappedPojo.class, csvFormat));
    pipeline.run();
  }