@Override
  protected @UnknownKeyFor @NonNull @Initialized SchemaTransform from(
      JdbcReadSchemaTransformConfiguration configuration) {
    configuration.validate();
    return new JdbcReadSchemaTransform(configuration);
  }
@Test
  public void testRead() {
    JdbcReadSchemaTransformProvider provider = null;
    for (SchemaTransformProvider p : ServiceLoader.load(SchemaTransformProvider.class)) {
      if (p instanceof JdbcReadSchemaTransformProvider) {
        provider = (JdbcReadSchemaTransformProvider) p;
        break;
      }
    }
    assertNotNull(provider);

    PCollection<Row> output =
        PCollectionRowTuple.empty(pipeline)
            .apply(
                provider.from(
                    JdbcReadSchemaTransformProvider.JdbcReadSchemaTransformConfiguration.builder()
                        .setDriverClassName(DATA_SOURCE_CONFIGURATION.getDriverClassName().get())
                        .setJdbcUrl(DATA_SOURCE_CONFIGURATION.getUrl().get())
                        .setLocation(READ_TABLE_NAME)
                        .build()))
            .get("output");
    Long expected = Long.valueOf(EXPECTED_ROW_COUNT);
    PAssert.that(output.apply(Count.globally())).containsInAnyOrder(expected);
    pipeline.run();
  }