@SuppressWarnings({
    "nullness" // TODO(https://github.com/apache/beam/issues/20497)
  })
  @Override
  protected SchemaTransform from(KafkaReadSchemaTransformConfiguration configuration) {
    return new KafkaReadSchemaTransform(configuration);
  }
@Test
  public void testBuildTransformWithAvroSchema() {
    ServiceLoader<SchemaTransformProvider> serviceLoader =
        ServiceLoader.load(SchemaTransformProvider.class);
    List<SchemaTransformProvider> providers =
        StreamSupport.stream(serviceLoader.spliterator(), false)
            .filter(provider -> provider.getClass() == KafkaReadSchemaTransformProvider.class)
            .collect(Collectors.toList());
    KafkaReadSchemaTransformProvider kafkaProvider =
        (KafkaReadSchemaTransformProvider) providers.get(0);
    kafkaProvider.from(
        KafkaReadSchemaTransformConfiguration.builder()
            .setFormat("AVRO")
            .setTopic("anytopic")
            .setBootstrapServers("anybootstrap")
            .setSchema(AVRO_SCHEMA)
            .build());
  }