public static JsonToRowWithErrFn withExceptionReporting(Schema rowSchema) {
    return JsonToRowWithErrFn.forSchema(rowSchema);
  }
@Test
  @Category(NeedsRunner.class)
  public void testParsesErrorWithErrorMsgWithRequireNullDeadLetter() 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)
                .withExtendedErrorInfo()
                .withNullBehavior(NullBehavior.REQUIRE_NULL));

    PCollection<Row> errorsWithMsg = results.getFailedToParseLines();

    PAssert.that(errorsWithMsg)
        .containsInAnyOrder(
            row(
                JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA,
                jsonPerson("person1", "80"),
                "Field 'knowsJavascript' is not present in the JSON object."),
            row(
                JsonToRowWithErrFn.ERROR_ROW_WITH_ERR_MSG_SCHEMA,
                jsonPerson("person3", "60"),
                "Field 'knowsJavascript' is not present in the JSON object."));

    pipeline.run();
  }