如何使用FlinkKafkaConsumer分别解析密钥<K，V>而不是<T>

"基于Flink Kafka Consumer，有一个构造函数：class KafkaRecord {
private K key;
private V value;
private long offset;
private int partition;
private String topic;

...
}

class MySchema implements KeyedDeserializationSchema> {
KafkaRecord deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) {

KafkaRecord<K, V> rec = new KafkaRecord<>();
rec.key = KEY_DESERIaLISER.deserialize(messageKey);
rec.value = ...;
rec.topic = topic;
...

}
}

public FlinkKafkaConsumer(String topic, KeyedDeserializationSchema deserializer, Properties props) {

this(Collections.singletonList(topic), deserializer, props);

}
第二个参数 - KeyedDeserializationSchema用于反序列化Kafka记录。它包括消息密钥，消息值，偏移量，主题等。因此，你可以MyKafkaRecord使用Avro密钥和Avro值实现名为T的自己的类型。然后，通过MyKafkaRecord为T您实现KeyedDeserializationSchema。请参考TypeInformationKeyValueSerializationSchema一个例子。

例如，阅读卡夫卡的额外信息："