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

    PCollection<Row> input =
        pipeline
            .apply(
                Create.of(
                    Row.withSchema(inputSchema).addValues(4).build(),
                    Row.withSchema(inputSchema).addValues(-1).build()))
            .setRowSchema(inputSchema);

    PCollectionRowTuple result =
        PCollectionRowTuple.of(JavaMapToFieldsTransformProvider.INPUT_ROWS_TAG, input)
            .apply(
                new JavaMapToFieldsTransformProvider()
                    .from(
                        JavaMapToFieldsTransformProvider.Configuration.builder()
                            .setLanguage("java")
                            .setFields(
                                ImmutableMap.of(
                                    "sqrt",
                                    JavaRowUdf.Configuration.builder()
                                        .setCallable(
                                            "import java.util.function.Function;"
                                                + "import org.apache.beam.sdk.values.Row;"
                                                + "public class Sqrt implements Function<Row, Double> {"
                                                + "  public Double apply(Row row) {"
                                                + "    int x = row.getInt32(\"x\");"
                                                + "    if (x < 0) {"
                                                + "      throw new ArithmeticException(\"negative value\");"
                                                + "    } else {"
                                                + "      return Math.sqrt(x);"
                                                + "    }"
                                                + "  }"
                                                + "}")
                                        .build()))
                            .setErrorHandling(ErrorHandling.builder().setOutput("errors").build())
                            .build()));

    PCollection<Row> sqrts = result.get(JavaMapToFieldsTransformProvider.OUTPUT_ROWS_TAG);
    Schema outputSchema = sqrts.getSchema();
    PAssert.that(sqrts)
        .containsInAnyOrder(Row.withSchema(outputSchema).withFieldValue("sqrt", 2.0).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(-1).build())
                .withFieldValue("error_message", "negative value")
                .build());
    pipeline.run();
  }