如何在单Kafka Connect JDBC Sink配置中为各主题指定pk.fields？

要在单个JDBC Sink连接器里给不同主题（对应数据库表）指定不同的主键字段，你可以借助Confluent的**Single Message Transform (SMT)**工具，通过Override转换器配合主题匹配规则来实现。下面是具体的配置方案：

核心思路 #

我们会添加多个Override转换步骤，每个步骤针对特定主题，动态覆盖pk.fields参数的值，同时保留原有的unwrap转换来处理Debezium的信封数据。

修改后的完整配置 #

jdbc-sink.source {
  "name": "jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "item,itemDetail,itemLocation",
    "connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=postgrespw",
    "transforms": "unwrap,overrideItemPk,overrideDetailPk,overrideLocationPk",
    "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
    # 针对item主题设置主键为id
    "transforms.overrideItemPk.type": "io.confluent.connect.transforms.Override$Config",
    "transforms.overrideItemPk.topic.regex": "item",
    "transforms.overrideItemPk.set": "pk.fields",
    "transforms.overrideItemPk.value": "id",
    # 针对itemDetail主题设置复合主键id,itemId
    "transforms.overrideDetailPk.type": "io.confluent.connect.transforms.Override$Config",
    "transforms.overrideDetailPk.topic.regex": "itemDetail",
    "transforms.overrideDetailPk.set": "pk.fields",
    "transforms.overrideDetailPk.value": "id,itemId",
    # 针对itemLocation主题设置复合主键id,itemId
    "transforms.overrideLocationPk.type": "io.confluent.connect.transforms.Override$Config",
    "transforms.overrideLocationPk.topic.regex": "itemLocation",
    "transforms.overrideLocationPk.set": "pk.fields",
    "transforms.overrideLocationPk.value": "id,itemId",
    "auto.create": "true",
    "insert.mode": "upsert",
    "pk.mode": "record_value"
  }
}

配置说明 #

• 新增的3个Override$Config类型转换，分别精准匹配item、itemDetail、itemLocation三个主题。
• 每个转换通过topic.regex锁定目标主题，用set指定要覆盖的配置项（这里是pk.fields），value则设置对应主题的主键字段。
• 保留原有的unwrap转换解析Debezium的事件信封，确保能获取到原始业务数据。
• pk.mode设为record_value，表示从消息的value中提取主键字段，和我们的配置逻辑完全兼容。

额外注意事项 #

• 确保你的Confluent Connect环境包含io.confluent.connect.transforms.Override这个SMT（Confluent Platform默认包中已自带）。
• 如果主题名称有更复杂的规则，可以调整topic.regex的匹配模式，比如用item.*匹配所有以item开头的主题。
• 复合主键的字段顺序要和数据库表中的主键顺序一致，避免upsert逻辑出错。

内容的提问来源于stack exchange，提问作者Shahid Ghafoor