Kafka JDBC Sink Connector如何在record_value为null时写入record_key字段

JDBC Sink连接器默认以Kafka记录的value作为写入数据库的数据源，你当前场景下所有记录的value为null，触发了连接器的非空校验；同时你原有配置中的InsertField转换使用错误，topic.field参数是用于将topic名称写入指定字段，不能实现读取key中字段的需求，所以运行报错。

修改连接器配置，通过内置转换将record key的内容映射为record value，替换原有null value后再写入数据库，修改后的完整配置如下：

{
  "name": "ObjectForDelete.sink",
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "transforms": [
    "hoistKey",
    "extractKey"
  ],
  "topics": [
    "ObjectForDeletePG"
  ],
  "transforms.hoistKey.type": "org.apache.kafka.connect.transforms.HoistField$Key",
  "transforms.hoistKey.field": "key_content",
  "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
  "transforms.extractKey.field": "key_content",
  "connection.url": "jdbc:postgresql://127.0.0.1:5432/ulk",
  "connection.user": "root",
  "connection.password": "****",
  "dialect.name": "PostgreSqlDatabaseDialect",
  "insert.mode": "insert",
  "table.name.format": "ObjectForDelete",
  "pk.mode": "none",
  "pk.fields": [],
  "db.timezone": "Europe/Kiev",
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "ObjectForDelete_sink_dlq",
  "errors.deadletterqueue.context.headers.enable": true
}

• hoistKey转换：将完整的record key包装为key结构体下的key_content临时字段
• extractKey转换：将key_content字段的内容提取为新的record value，直接替换原有null value，此时连接器就可以从新的value（也就是原record key）中读取uuid、name字段写入数据库
• 错误容忍相关配置为可选配置：如果需要跳过异常记录、避免单条异常数据导致连接器停止运行可以保留，也可以根据实际需求删除
• 若原record key中的字段名和数据库表字段名不一致，可以额外新增ReplaceField类型的转换完成字段重命名

内容的提问来源于stack exchange，提问作者Леонид Дубравский