带有kafka-avro-console-consumer的未知魔术字节

Question

我一直在尝试将kafka-avro-console-consumer从Confluent连接到我们的遗留Kafka集群，该集群是在没有Confluent模式注册表的情况下部署的。我使用以下属性显式提供了模式：

kafka-console-consumer --bootstrap-server kafka02.internal:9092 \
    --topic test \
    --from-beginning \
    --property key.schema='{"type":"long"}' \
    --property value.schema='{"type":"long"}'

但我得到的是‘未知的魔法字节！’org.apache.kafka.common.errors.serializationexception出错

是否可以使用Confluent kafka-avro-console-consumer（未使用Confluent的AvroSerializer序列化）和Schema Registry来使用来自Kafka的Avro消息？

Answer 1

汇合模式注册表序列化器/反序列化器使用有线格式，该格式在消息的初始字节中包含关于模式ID等的信息。

如果您的消息没有使用架构注册表序列化程序序列化，那么您将无法用它反序列化它，并且将得到unknown magic byte！错误。

因此，您需要编写一个使用者来提取消息，使用Avro avsc模式执行反序列化，然后假设您希望保留数据，使用模式注册表序列化器重新序列化它

编辑：我最近写了一篇文章，更深入地解释了这整件事：https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explaned