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

{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "int"}
  ]
}

import org.apache.flink.formats.avro.AvroDeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.avro.specific.SpecificRecord;

public class UserAvroDeserializationSchema extends AvroDeserializationSchema<User> {

    public UserAvroDeserializationSchema(Class<User> type) {
        super(type);
    }

    @Override
    public User deserialize(byte[] bytes) {
        // 反序列化Avro数据为User对象
        User user = new User();
        // ...
        return user;
    }

    @Override
    public TypeInformation<User> getProducedType() {
        return TypeInformation.of(User.class);
    }
}

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.api.common.serialization.SimpleStringSchema;

public class KafkaAvroConsumer {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");

        FlinkKafkaConsumer<User> consumer = new FlinkKafkaConsumer<>("topic", new UserAvroDeserializationSchema(User.class), properties);

        env.addSource(consumer)
           .print();

        env.execute("Kafka Avro Consumer");
    }
}