我尝试使用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连接器，并与控制台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格式作为输入。
我觉得接收器连接器中avro格式的反序列化没有正确实现，因为控制台avro使用者仍然可以成功读取数据。有人知道这种行为的原因吗？我也希望能有一个简单的解决办法！