Kafka sink 连接器中的 Avro 反序列化问题答案

【问题标题】：Problems with Avro deserialization in Kafka sink connectorsKafka sink 连接器中的 Avro 反序列化问题
【发布时间】：2019-03-03 08:56:39
【问题描述】：

我正在尝试使用 Kafka 从 DB2 读取数据，然后将其写入 HDFS。我使用带有标准 JDBC 和 HDFS 连接器的分布式融合平台。由于 HDFS 连接器需要知道模式，因此它需要 avro 数据作为输入。因此，我必须为馈送到 Kafka 的数据指定以下 avro 转换器（在 etc/kafka/connect-distributed.properties 中）：

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

然后我运行我的 JDBC 连接器并使用 console-avro-consumer 检查我是否可以成功读取从 DB2 获取的数据。

但是，当我启动 HDFS 连接器时，它不再工作了。相反，它输出 SerializationException:

Error deserializing Avro message for id -1
... Unknown magic byte!

为了检查这是否是 HDFS 连接器的问题，我尝试使用简单的 FileSink 连接器。但是，我在使用 FileSink 时看到了完全相同的异常（并且文件本身已创建但保持为空）。

然后我进行了以下实验：我没有使用 avro 转换器作为键和值，而是使用了 json 转换器：

key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schema.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schema.enable=false

这解决了 FileSink 连接器的问题，即从 DB2 到文件的整个管道工作正常。但是，对于 HDFS 连接器，此解决方案是不可行的，因为连接器需要架构，因此需要 avro 格式作为输入。

在我看来，sink 连接器中 avro 格式的反序列化没有正确实现，因为 console-avro-consumer 仍然可以成功读取数据。有谁知道这种行为的原因是什么？我也很感激一个简单的解决方法！

【问题讨论】：

标签： apache-kafka hdfs avro confluent-platform confluent-schema-registry

【解决方案1】：

检查 console-avro-consumer 是否可以成功读取获取的数据

我猜你这样做时没有添加--property print.key=true --from-beginning。

最新的值可能是 Avro，但 connect 显然在该主题的某个地方失败，因此您需要扫描它以找出发生这种情况的位置

如果使用JsonConverter 有效，并且数据实际上是磁盘上可读的 JSON，那么听起来 JDBC 连接器实际上写的是 JSON，而不是 Avro

如果您能够查明错误消息的偏移量，则可以使用设置了连接器组 ID 的常规控制台使用者，然后添加 --max-messages 以及指定的分区和偏移量以跳过这些事件

【讨论】：

我也尝试添加 --property print.key=true 并且我看到该键为 null。之后我尝试使用 StringConverter 但它没有改变任何东西 - 我得到了完全相同的异常。
你从话题开始就开始消费了吗？您确定您使用的是 Avro 消费者吗？因为您的错误来自控制台使用者和 Connect 中的 AvroConverter 共享的反序列化器，所以您应该看到同样的问题
这将从最新的当前偏移量开始消耗，直到您添加 --from-beginning，但我不希望您的数据在主题中更改类型，除非有人在主题中生成随机消息（通过错误？）
这是一个很好的观点，谢谢！我尝试了使用 console-avro-consumer 的--from-beginning 选项，它突然崩溃，出现与我在使用 hdfs 或文件接收器时看到的相同的 avro 反序列化异常（未知魔术字节）。这让我很困惑，因为数据来自同一个表（我只是在我的 JDBC 连接器中针对表指定了一个选择查询）。你知道为什么在这种情况下数据会改变它的类型吗？这可能是由于缺少数据或其他原因造成的吗？
我猜有人在某个地方的主题上生成了一个常规字符串...您可以使用常规控制台使用者，然后查看看起来不像其他消息的消息