public static JsonToRowWithErrFn withExceptionReporting(Schema rowSchema) {
    return JsonToRowWithErrFn.forSchema(rowSchema);
  }
@Test
  @Category(NeedsRunner.class)
  public void testParsesErrorWithErrorMsgRowsDeadLetterWithCustomFieldNames() throws Exception {
    PCollection<String> jsonPersons =
        pipeline.apply("jsonPersons", Create.of(JSON_PERSON_WITH_ERR));

    String customErrField = "CUSTOM_ERR";
    String customLineField = "CUSTOM_LINE";

    ParseResult results =
        jsonPersons.apply(
            JsonToRow.withExceptionReporting(PERSON_SCHEMA)
                .withExtendedErrorInfo()
                .setErrorField(customErrField)
                .setLineField(customLineField));

    Schema customSchema =
        Schema.builder()
            .addField(customLineField, FieldType.STRING)
            .addField(customErrField, FieldType.STRING)
            .build();

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

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

    PAssert.that(errorsWithMsg)
        .containsInAnyOrder(
            row(customSchema, "{}", "Non-nullable field 'name' is not present in the JSON object."),
            row(customSchema, "Is it 42?", "Unable to parse Row"));

    pipeline.run();
  }