java - 使用 KafkaAvroDecoder 将 Avro 消息反序列化为特定数据

Question

我正在阅读 Kafka 主题，其中包含使用 KafkaAvroEncoder(自动将模式注册到主题)序列化的 Avro 消息。我正在使用 maven-avro-plugin 生成纯 Java 类，我想在阅读时使用它。

KafkaAvroDecoder 仅支持反序列化为 GenericData.Record 类型，这(在我看来)忽略了静态类型语言的全部意义。我的反序列化代码目前看起来像这样:

    SpecificDatumReader<event> reader = new SpecificDatumReader<>(
        event.getClassSchema() // event is my class generated from the schema
    );
    byte[] in = ...; // my input bytes;
    ByteBuffer stuff = ByteBuffer.wrap(in);
    // the KafkaAvroEncoder puts a magic byte and the ID of the schema (as stored 
    //   in the schema-registry) before the serialized message
    if (stuff.get() != 0x0) {
        return;
    }
    int id = stuff.getInt();

    // lets just ignore those special bytes
    int length = stuff.limit() - 4 - 1;
    int start = stuff.position() + stuff.arrayOffset();

    Decoder decoder = DecoderFactory.get().binaryDecoder(
        stuff.array(), start, length, null
    );
    try {
        event ev = reader.read(null, decoder);
    } catch (IOException e) {
        e.printStackTrace();
    }

我发现我的解决方案很麻烦，所以我想知道是否有更简单的解决方案。

Answer 1

多亏了评论，我才能找到答案。秘诀是实例化 KafkaAvroDecoder用Properties指定特定 Avro 阅读器的使用，即:

    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "...");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
            io.confluent.kafka.serializers.KafkaAvroSerializer.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_C‌ONFIG, "...");
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
    VerifiableProp vProps = new VerifiableProperties(props);

    KafkaAvroDecoder decoder = new KafkaAvroDecoder(vProps);
    MyLittleData data = (MyLittleData) decoder.fromBytes(input);

同样的配置适用于直接使用KafkaConsumer<K, V>的情况。类(我在 Storm 中使用来自 storm-kafka 项目的 KafkaSpout 消费，它使用 SimpleConsumer ，所以我必须手动反序列化消息。对于勇敢的人来说，有 storm-kafka-client 项目，它通过使用新样式的消费者自动执行此操作)。