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

    ParseResult results =
        jsonPersons.apply(JsonToRow.withExceptionReporting(PERSON_SCHEMA).withExtendedErrorInfo());

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

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

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

    pipeline.run();
  }