@Override
  protected SchemaTransform from(Configuration configuration) {
    return new JavaFilterTransform(configuration);
  }
@Test
  @Category(NeedsRunner.class)
  public void testErrorHandling() {
    Schema inputSchema = Schema.of(Schema.Field.of("s", Schema.FieldType.STRING));

    PCollection<Row> input =
        pipeline
            .apply(
                Create.of(
                    Row.withSchema(inputSchema).addValues("short").build(),
                    Row.withSchema(inputSchema).addValues("looooooooooooong").build()))
            .setRowSchema(inputSchema);

    PCollectionRowTuple result =
        PCollectionRowTuple.of(JavaFilterTransformProvider.INPUT_ROWS_TAG, input)
            .apply(
                new JavaFilterTransformProvider()
                    .from(
                        JavaFilterTransformProvider.Configuration.builder()
                            .setLanguage("java")
                            .setKeep(
                                JavaRowUdf.Configuration.builder()
                                    .setExpression("s.charAt(7) == 'o'")
                                    .build())
                            .setErrorHandling(ErrorHandling.builder().setOutput("errors").build())
                            .build()));

    PCollection<Row> good = result.get(JavaFilterTransformProvider.OUTPUT_ROWS_TAG);
    PAssert.that(good)
        .containsInAnyOrder(
            Row.withSchema(inputSchema).withFieldValue("s", "looooooooooooong").build());

    PCollection<Row> errors = result.get("errors");
    Schema errorSchema = errors.getSchema();
    PAssert.that(errors)
        .containsInAnyOrder(
            Row.withSchema(errorSchema)
                .withFieldValue(
                    "failed_row", Row.withSchema(inputSchema).addValues("short").build())
                .withFieldValue("error_message", "String index out of range: 7")
                .build());
    pipeline.run();
  }