如何创建AvroDeserialzationSchema并在Flink Kafka Consumer中使用？

我挖掘了kafka消费者javadocs并找到了一个消耗avro流的例子。我仍然需要将kafka消费转换为flinkKafkaConsumer，但下面的代码有效。

对于io.confluent引用工作，我必须添加存储库和pom文件的依赖项。

confluent

<url>http://packages.confluent.io/maven/</url>

<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>3.1.1</version>

public class StreamingJob {

// static DeserializationSchema avroSchema = new AvroDeserializationSchema(pendingsv.class);

public static void main(String[] args) throws Exception {
    // set up the streaming execution environment
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "opssupport.alarms");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");
    props.put("schema.registry.url", "http://localhost:8081");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    String topic = "pendingSVs_";
    final Consumer<String, GenericRecord> consumer = new KafkaConsumer<String, GenericRecord>(props);
    consumer.subscribe(Arrays.asList(topic));

    try {
        while (true) {
            ConsumerRecords<String, GenericRecord> records = consumer.poll(100);
            for (ConsumerRecord<String, GenericRecord> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value());
            }
        }
    } finally {
        consumer.close();
    }

// execute program
//env.execute("Flink Streaming Java API Skeleton");

}