Confluent Kafka Rest Proxy - Avro 反序列化答案

【问题标题】：Confluent Kafka Rest Proxy - Avro DeserializationConfluent Kafka Rest Proxy - Avro 反序列化
【发布时间】：2019-07-30 10:07:00
【问题描述】：

我正在尝试使用 Confluent Kafka REST 代理从我的一个主题中检索 Avro 格式的数据，但不幸的是我收到了反序列化错误。我正在使用以下命令查询 Kafka REST 代理

 curl -X GET -H "Accept: application/vnd.kafka.avro.v2+json" 
http://localhost:8082/consumers/my-group/instances/my-consumer/records?timeout=30000

我得到了回应

{
  "error_code": 50002,
  "message": "Kafka error: Error deserializing key/value for partition input-0 at offset 0. If needed, please seek past the record to continue consumption."
}

Kafka Rest Proxy 服务器上的日志是：

org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition input-0 at offset 0. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

数据已使用 KafkaAvroSerializer 生成，并且架构存在于架构注册表中。另请注意，在 CLI 上使用 avro-console-consumer 可以读取数据。

有人知道如何解决这个问题吗？

【问题讨论】：

标签： apache-kafka confluent-platform kafka-rest

【解决方案1】：

很可能除了关于该主题的有效 Avro 消息外，您也有无效消息。这就是这个错误的含义，也正是我尝试使用 REST 代理在本地使用非 Avro 消息时遇到的错误：

ERROR Unexpected exception in consumer read task id=io.confluent.kafkarest.v2.KafkaConsumerReadTask@2e20d4f3  (io.confluent.kafkarest.v2.KafkaConsumerReadTask)
org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition avrotest-0 at offset 2. If needed, please seek past the record to continue consumption.
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id -1
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

我会使用kafkacat之类的工具来检查错误中给出的偏移量处的实际消息，例如：

kafkacat -C -b localhost:9092 -t test_topic_avro -o 0 -c 1

-o 0 将在偏移量 0 处消费消息，-c 1 表示只消费一条消息。

您也可以seek past the problematic offset，例如对于主题avrotest，将偏移量移动到1：

echo '{ "offsets": [ { "topic": "avrotest", "partition": 0, "offset": 1 } ] }' | \
http POST localhost:8082/consumers/rmoff_consumer_group/instances/rmoff_consumer_instance/positions \
Content-Type:application/vnd.kafka.v2+json

【讨论】：

好的，我可以提交一条消息并继续下一条，但问题仍然存在。另一方面，我也可以使用 kafka-topics-ui 阅读消息。问题可能是我对键（StringSerializer）和值（KafkaAvroSerializer）使用了不同的序列化程序吗？

【解决方案2】：

直到最近才支持在 rest 代理中包含字符串键和 AVRO 值：

https://github.com/confluentinc/kafka-rest/issues/210

最近代码已合并，但问题仍未解决，文档尚未完全更新：

https://github.com/confluentinc/kafka-rest/pull/797

【讨论】：