如何将参数传递给apache (KafkaIO)中的avro反序列化程序？

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.kafka.KafkaIO;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.KV;
import org.apache.kafka.common.serialization.StringDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;

public class KafkaAvroExample {
    public static void main(String[] args) {
        Pipeline pipeline = Pipeline.create();

        pipeline.apply(KafkaIO.<String, String>read()
            .withBootstrapServers("kafka-server:9092")
            .withTopic("input-topic")
            .withKeyDeserializer(StringDeserializer.class)
            .withValueDeserializer(new KafkaAvroDeserializer())
            .withSchemaRegistryUrl("http://schema-registry:8081")
            .withoutMetadata() // 可选，移除元数据
            .updateConsumerProperties(ImmutableMap.of(
                "specific.avro.reader", "true" // 传递参数示例
            ))
        )
        .apply(ParDo.of(new DoFn<KV<String, String>, Void>() {
            @ProcessElement
            public void processElement(ProcessContext c) {
                // 处理反序列化后的数据
            }
        }));

        pipeline.run().waitUntilFinish();
    }
}