public static JsonToRowWithErrFn withExceptionReporting(Schema rowSchema) {
    return JsonToRowWithErrFn.forSchema(rowSchema);
  }
@Test
  @Category(NeedsRunner.class)
  public void testParsesRowsDeadLetterWithMissingFieldsNoErrors() throws Exception {

    PCollection<String> jsonPersons =
        pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_IMPLICIT_NULLS));

    ParseResult results =
        jsonPersons.apply(
            JsonToRow.withExceptionReporting(PERSON_SCHEMA_WITH_NULLABLE_FIELD)
                .withNullBehavior(NullBehavior.ACCEPT_MISSING_OR_NULL));

    PCollection<Row> personRows = results.getResults();
    PCollection<Row> errors = results.getFailedToParseLines();

    PAssert.that(personRows).containsInAnyOrder(PERSON_ROWS_WITH_NULLS);

    PAssert.that(errors).empty();

    pipeline.run();
  }