Confluent S3 Sink连接器中消息Key的存储位置咨询

我来帮你理清这个问题哈！Confluent的S3 Sink连接器默认情况下确实只会把消息的Value部分序列化到Avro文件中，Key不会自动被包含进去，这就是你只看到消息本身的原因。下面给你几种常见的处理方案：

1. 单独存储消息Key #

你可以通过修改连接器配置，让它单独存储消息Key：

• 添加配置项 store.kafka.keys=true，开启Key存储功能
• 同时需要配置Key的转换器，和Value保持一致（比如用Avro的话）：

  key.converter=io.confluent.connect.avro.AvroConverter
  key.converter.schema.registry.url=http://你的Schema Registry地址:8081
  

开启后，连接器会在S3的对应路径下生成单独的Key文件（通常命名会包含-keys后缀），和Value文件一一对应，你可以导出这些Key文件来获取消息的Key内容。

2. 将Key嵌入到Value中（同文件存储） #

如果希望Key和Value存放在同一个Avro文件里，你可以使用**Single Message Transform（SMT）**来把Key插入到Value的字段中：

• 添加以下SMT配置，把Key插入到Value的message_key字段里（字段名可以自定义）：

  transforms=insertKey
  transforms.insertKey.type=org.apache.kafka.connect.transforms.InsertField$Value
  transforms.insertKey.field=message_key
  transforms.insertKey.static.field=${kafka.key}
  

这样处理后，导出的Avro文件里每条记录都会包含你指定的message_key字段，直接就能看到对应的Key内容了。

额外注意点 #

• 如果你之前已经运行过连接器，修改配置后需要重启连接器才能生效
• 若使用Schema Registry，确保Key的Schema已经在Registry中注册（或者开启自动注册：key.converter.schema.registry.auto.register=true）

内容的提问来源于stack exchange，提问作者Xiang Zhang