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

Question

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

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 消费来自 Kafka 的 Avro 消息，这些消息未使用 Confluent 的 AvroSerializer 和 Schema Registry 序列化？

Answer 1

Confluent Schema Registry 序列化器/反序列化器使用一种有线格式，它在消息的初始字节中包含有关模式 ID 等的信息。

如果您的消息尚未使用 Schema Registry 序列化程序进行序列化，那么您将无法使用它反序列化它，并且会收到Unknown magic byte!错误消息。

因此，您需要编写一个消费者来提取消息，使用您的 Avro avsc 模式进行反序列化，然后假设您想保留数据，使用Schema Registry 序列化程序重新序列化它

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