无法阅读Kafka主题avro消息

Question

Debezium连接器的Kafka connect事件是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-独立-使用者使用事件，但它也抛出了一个数组索引。因此，我的猜测是消息在Kafka连接（生产者）上编码不正确

以下是问题：

1. 在生产者或消费者处传递的配置是否有问题？
2. 为什么键反序列化有效，但价值反序列化不起作用？
3. 要使事情发挥作用，还需要做些什么吗？（如在某处指定字符编码）。
4. 带有Avro的Debezium是否可以用于生产，还是现在它是一个实验性功能？关于德贝兹乌姆·阿夫罗的帖子特别指出，涉及阿夫罗的例子将在未来被包括在内。

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

Answer 1

按照http://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html的步骤