配置Kafka S3 Json Connector后为何出现Avro转换异常？

我来帮你拆解这个问题——你遇到的核心矛盾其实是消息反序列化逻辑和S3写入格式的混淆：虽然你指定了S3文件用JSON格式，但Kafka Connect在从Kafka主题读取消息时，依然在使用Avro转换器解析数据，而你发送的是普通JSON字符串，这就导致了序列化异常。

问题根源解析 #

• Kafka Connect的转换器（Converter）和S3 Sink的format.class是完全独立的两个配置：
  • format.class只负责控制最终写入S3的文件格式是JSON还是其他类型；
  • 而转换器（key.converter/value.converter）负责把Kafka主题里的字节数据转换成Connect能处理的结构化数据。
• 默认情况下，Confluent Platform的Kafka Connect全局配置会把value.converter设为io.confluent.connect.avro.AvroConverter，这个转换器要求消息必须是Avro序列化格式（开头有特定的magic字节，还要关联Schema Registry的schema ID）。但你用kafka-console-producer发送的是纯JSON字符串，没有Avro的标识，所以转换器尝试解析时就会抛出Unknown magic byte!和Error deserializing Avro message for id -1的错误。

解决方案 #

你需要在S3 Sink连接器的配置里显式指定JSON转换器，告诉Connect用正确的逻辑解析你的消息，具体步骤如下：

1. 更新连接器配置，添加以下两个关键参数：

   • value.converter=org.apache.kafka.connect.json.JsonConverter：指定用JSON转换器处理消息体
   • value.converter.schemas.enable=false：禁用schema检查（因为你发送的JSON没有包含schema信息）

   更新后的完整配置示例：

   {
     "name": "s3-sink",
     "config": {
       "connector.class": "io.confluent.connect.s3.S3SinkConnector",
       "tasks.max": "1",
       "topics": "s3_hose",
       "s3.region": "us-east-1",
       "s3.bucket.name": "some-bucket-name",
       "s3.part.size": "5242880",
       "flush.size": "1",
       "storage.class": "io.confluent.connect.s3.storage.S3Storage",
       "format.class": "io.confluent.connect.s3.format.json.JsonFormat",
       "schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
       "partitioner.class": "io.confluent.connect.storage.partitioner.DefaultPartitioner",
       "schema.compatibility": "NONE",
       "value.converter": "org.apache.kafka.connect.json.JsonConverter",
       "value.converter.schemas.enable": "false",
       "name": "s3-sink"
     },
     "tasks": [{"connector": "s3-sink", "task": 0}],
     "type": null
   }
   

2. 重新加载连接器：

   # 先卸载旧的连接器实例
   confluent unload s3-sink
   # 加载更新后的配置
   confluent load s3-sink
   

3. 重新发送测试数据：

   kafka-console-producer --broker-list localhost:9092 --topic s3_hose
   # 输入 {"q":1} 并回车
   

验证结果 #

此时再查看连接器日志，Avro相关的异常应该会消失，你也能在指定的S3 Bucket里找到包含{"q":1}内容的JSON文件。

补充说明：如果你的Kafka Connect全局配置已经把value.converter设为JSON转换器，那连接器级的配置可以省略，但为了连接器的独立性，建议显式指定，避免受全局配置变更影响。

内容的提问来源于stack exchange，提问作者Vova l