是否有可能反序列化Avro消息（消耗来自Kafka的消息），而不给Reader模式

Question

我正在使用Apache Flink中的Kafka连接器来访问合流Kafka提供的流。

除了模式注册表urlConfluentRegistryAvroDeserializationSchema之外。伪造的（…） 应为“reader”架构。我不想提供读取模式，而是想使用同一个编写器的模式（在注册表中查找）来读取消息，因为消费者不会有最新的模式。

FlinkKafkaConsumer010<GenericRecord> myConsumer =
        new FlinkKafkaConsumer010<>("topic-name", ConfluentRegistryAvroDeserializationSchema.forGeneric(<reader schema goes here>, "http://host:port"), properties);
myConsumer.setStartFromLatest();

https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html"使用这些反序列化模式记录将使用从模式注册表检索并转换为静态提供的模式读取"

既然我不想在消费者端保留模式定义，我该如何使用writer的模式反序列化来自Kafka的Avro消息？

感谢您的帮助！




Answer 1

我认为不可能直接使用ConFluentCORstryAvroDeseriazationSchema.forGeneric。它旨在与读取器模式一起使用，并且它们对此有先决条件检查。

你必须实现你自己的。两个进口的东西：

• 设置specific。阿夫罗。reader为false（否则你会得到具体的记录）
• KafkaAvroDeserializer必须延迟初始化（因为它本身不可序列化，因为它包含对schema注册表客户端的引用）

import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
import java.util.HashMap;
import java.util.Map;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;

public class KafkaGenericAvroDeserializationSchema
    implements KeyedDeserializationSchema<GenericRecord> {

  private final String registryUrl;
  private transient KafkaAvroDeserializer inner;

  public KafkaGenericAvroDeserializationSchema(String registryUrl) {
    this.registryUrl = registryUrl;
  }

  @Override
  public GenericRecord deserialize(
      byte[] messageKey, byte[] message, String topic, int partition, long offset) {
    checkInitialized();
    return (GenericRecord) inner.deserialize(topic, message);
  }

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

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

  private void checkInitialized() {
    if (inner == null) {
      Map<String, Object> props = new HashMap<>();
      props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, registryUrl);
      props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, false);
      SchemaRegistryClient client =
          new CachedSchemaRegistryClient(
              registryUrl, AbstractKafkaAvroSerDeConfig.MAX_SCHEMAS_PER_SUBJECT_DEFAULT);
      inner = new KafkaAvroDeserializer(client, props);
    }
  }
}

env.addSource(
  new FlinkKafkaConsumer<>(
    topic, 
    new KafkaGenericAvroDeserializationSchema(schemaReigstryUrl), 
    kafkaProperties));