@Override
  protected SchemaTransform from(FileWriteSchemaTransformConfiguration configuration) {
    return new FileWriteSchemaTransform(configuration);
  }
@Test
  public void receivedUnexpectedInputTagsThrowsAnError() {
    SchemaTransform transform = PROVIDER.from(defaultConfiguration().setFormat(JSON).build());
    PCollectionRowTuple empty = PCollectionRowTuple.empty(errorPipeline);
    IllegalArgumentException emptyInputError =
        assertThrows(IllegalArgumentException.class, () -> empty.apply(transform));

    assertEquals(
        "org.apache.beam.sdk.io.fileschematransform.FileWriteSchemaTransformProvider$FileWriteSchemaTransform expects a single input tagged PCollection<Row> input",
        emptyInputError.getMessage());

    PCollection<Row> rows1 =
        errorPipeline.apply(
            Create.of(DATA.allPrimitiveDataTypesRows)
                .withRowSchema(ALL_PRIMITIVE_DATA_TYPES_SCHEMA));
    PCollection<Row> rows2 =
        errorPipeline.apply(
            Create.of(DATA.timeContainingRows).withRowSchema(TIME_CONTAINING_SCHEMA));

    PCollectionRowTuple tooManyTags =
        PCollectionRowTuple.of(INPUT_TAG, rows1).and("another", rows2);
    IllegalArgumentException tooManyTagsError =
        assertThrows(IllegalArgumentException.class, () -> tooManyTags.apply(transform));

    assertEquals(
        "org.apache.beam.sdk.io.fileschematransform.FileWriteSchemaTransformProvider$FileWriteSchemaTransform expects a single input tagged PCollection<Row> input",
        tooManyTagsError.getMessage());

    // should not throw an error
    PCollectionRowTuple input = PCollectionRowTuple.of(INPUT_TAG, rows1);
    input.apply(transform);
  }