无法读取 Kafka 主题 avro 消息

Question

Debezium 连接器的 Kafka 连接事件是 Avro 编码的。

在传递给 Kafka connect standalone 服务的 connect-standalone.properties 中提到了以下内容。

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

使用这些属性配置 Kafka 消费者代码：

Properties props = new Properties();
props.put("bootstrap.servers", "ip_address:9092");
props.put("zookeeper.connect", "ip_address:2181");
props.put("group.id", "test-consumer-group");
props.put("auto.offset.reset","smallest");
//Setting auto comit to false to ensure that on processing failure we retry the read
props.put("auto.commit.offset", "false");
props.put("key.converter.schema.registry.url", "ip_address:8081");
props.put("value.converter.schema.registry.url", "ip_address:8081");
props.put("schema.registry.url", "ip_address:8081");

在消费者实现中，以下是读取键和值组件的代码。我正在使用 REST 从模式注册表获取键和值的模式。

GenericDatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>(schema);
return reader.read(null, DecoderFactory.get().binaryDecoder(byteData, null));

解析密钥工作正常。在解析消息的值部分时，我收到 ArrayIndexOutOfBoundsException。

下载Avro的源代码并调试。发现 GenericDatumReader.readInt 方法返回负值。该值应该是数组（符号）的索引，因此应该是正数。

尝试使用 kafka-avro-standalone-consumer 消费事件，但它也抛出了 ArrayIndexOutOfBoundsException。所以，我的猜测是消息在 Kafka connect（生产者）上编码不正确，问题出在配置上。

问题如下：

生产者或消费者传递的配置是否有问题？
为什么密钥反序列化有效但价值无效？
是否需要做任何其他事情才能正常工作？（比如在某处指定字符编码）。
带有 Avro 的 Debezium 可以在生产中使用吗，还是目前是实验性功能？ Debezium Avro 上的 post 明确表示将来会包含涉及 Avro 的示例。

有很多 post Avro 反序列化抛出 ArrayIndexOutOfBoundsException 但无法将其与我面临的问题联系起来。

Answer 1

已按照 http://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html 中的步骤进行操作，现在一切正常。

无法读取 Kafka 主题 avro 消息

Unable to read Kafka topic avro messages

apache-kafka

avro

debezium