在Flink 1.9.1下使用confluent注册表序列化Kafka消息

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

import io.confluent.kafka.serializers.KafkaAvroDeserializationSchema;
import io.confluent.kafka.serializers.KafkaAvroSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

// 创建一个Avro的数据模型
public class MyAvroRecord {
    private String field1;
    private int field2;
    
    // getters and setters
}

// 使用KafkaAvroDeserializationSchema进行反序列化
KafkaAvroDeserializationSchema<MyAvroRecord> deserializationSchema = new KafkaAvroDeserializationSchema<>(MyAvroRecord.class);

// 使用KafkaAvroSerializationSchema进行序列化
KafkaAvroSerializationSchema<MyAvroRecord> serializationSchema = new KafkaAvroSerializationSchema<>(topic, schemaRegistryUrl);

// 创建Flink Kafka Consumer和Producer
FlinkKafkaConsumer<MyAvroRecord> kafkaConsumer = new FlinkKafkaConsumer<>(topic, deserializationSchema, properties);
FlinkKafkaProducer<MyAvroRecord> kafkaProducer = new FlinkKafkaProducer<>(topic, serializationSchema, properties);

properties.setProperty("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092");
properties.setProperty("group.id", "flink-consumer-group");
properties.setProperty("schema.registry.url", "http://schema-registry:8081");

DataStream<MyAvroRecord> stream = env.addSource(kafkaConsumer);
stream.map(record -> {
    // 对消息进行处理
    return record;
}).addSink(kafkaProducer);