我也在做类似的事情（我用的是09消费者）
在自定义反序列化程序的主代码传递中：

FlinkKafkaConsumer09<Object> kafkaConsumer = new FlinkKafkaConsumer09<>(
                parameterTool.getRequired("topic"), new MyDeserializationSchema<>(),
                parameterTool.getProperties());

自定义反序列化模式读取字节，找出模式并/或从模式注册表检索它，反序列化为genericrecord并返回genericrecord对象。

public class MyDeserializationSchema<T> implements DeserializationSchema<T> {

    private final Class<T> avrotype = (Class<T>) org.apache.avro.generic.GenericRecord.class;

    @Override
    public T deserialize(byte[] arg0) throws IOException {
        //do your stuff here, strip off your bytes
        //deserialize and create your GenericRecord 
        return (T) (myavroevent);
    }

    @Override
    public boolean isEndOfStream(T nextElement) {
        return false;
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return TypeExtractor.getForClass(avrotype);
    }

}

如果您使用confluent的schema registry，我相信首选的解决方案是使用confluent提供的avro serde。这样，我们就打电话 deserialize() 而要使用的最新版本avro模式的分辨率是在后台自动完成的，不需要字节操作。
它可以归结为以下内容（scala中的示例代码，java解决方案非常类似）：

import io.confluent.kafka.serializers.KafkaAvroDeserializer

...

val valueDeserializer = new KafkaAvroDeserializer()
valueDeserializer.configure(
  Map(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG -> schemaRegistryUrl).asJava, 
  false)

...

override def deserialize(messageKey: Array[Byte], message: Array[Byte], 
                       topic: String, partition: Int, offset: Long): KafkaKV = {

    val key = keyDeserializer.deserialize(topic, messageKey).asInstanceOf[GenericRecord]
    val value = valueDeserializer.deserialize(topic, message).asInstanceOf[GenericRecord]

    KafkaKV(key, value)
    }

...

此方法要求消息生产者也与模式注册表集成，并在那里发布模式。这可以用与上面非常相似的方法来完成，使用confluent的 KafkaAvroSerializer 我在这里发布了一个详细的解释：如何将flink与confluent的模式注册表集成