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

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

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

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

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

    PAssert.that(errors)
        .containsInAnyOrder(
            row(JsonToRowWithErrFn.ERROR_ROW_SCHEMA, "{}"),
            row(JsonToRow.JsonToRowWithErrFn.ERROR_ROW_SCHEMA, "Is it 42?"));

    pipeline.run();
  }