@SuppressWarnings({
    "nullness" // TODO(https://github.com/apache/beam/issues/20497)
  })
  @Override
  protected SchemaTransform from(KafkaReadSchemaTransformConfiguration configuration) {
    return new KafkaReadSchemaTransform(configuration);
  }
@Test
  public void testBuildTransformWithProtoFormat() {
    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()
            .setTopic("anytopic")
            .setBootstrapServers("anybootstrap")
            .setFormat("PROTO")
            .setMessageName("MyMessage")
            .setFileDescriptorPath(
                Objects.requireNonNull(
                        getClass().getResource("/proto_byte/file_descriptor/proto_byte_utils.pb"))
                    .getPath())
            .build());
  }