Debezium Sink连接器同步PostgreSQL CDC数据报错求助

问题根源 #

1. CDC消息未解析：Kafka中的消息是Debezium标准的Envelope结构（包含before/after/op等元数据），当前Sink直接消费整个结构体，无法匹配目标表的业务字段。
2. 主键与表结构配置冲突：primary.key.mode: none导致连接器无法识别目标表主键，加上schema.evolution: basic会尝试自动修改表结构，但目标表主键字段非空且无默认值，触发ALTER表失败。
3. 插入模式不兼容：Kafka中存在更新操作（op: u），但insert.mode: insert仅支持插入，无法处理更新，导致数据写入中断。

分步解决方案 #

1. 提取业务数据（必做）

添加Debezium的ExtractNewRecordState转换，从CDC信封中提取after字段的实际业务数据，同时保留操作类型以适配增删改：

"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite"

2. 修正核心配置

• 主键配置：设置primary.key.mode: record_value并指定实际主键字段，让连接器识别目标表主键，避免不必要的表结构修改。
• 插入模式：改为upsert，支持插入、更新、删除操作（适配CDC的c/u/d操作类型）。
• 表结构演化：因为目标表已提前创建，将schema.evolution设为none，禁止连接器自动修改表结构。
• 表名映射：如果topic名称是public.[TableName]，调整table.name.format去掉前缀，匹配目标表名：

  "table.name.format": "public.${topic.replace('public.', '')}"
  

最终可用配置

{
    "name": "sink-connector",
    "config": {
        "connector.class": "io.debezium.connector.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics.regex": "public.*",
        "connection.url": "jdbc:postgresql://postgres:5432/db",
        "connection.username": "postgres",
        "connection.password": "postgres",
        "insert.mode": "upsert",
        "table.name.format": "public.${topic.replace('public.', '')}",
        "primary.key.mode": "record_value",
        "primary.key.fields": "Column1", // 替换为你的实际主键字段，多个用逗号分隔
        "schema.evolution": "none",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "true",
        "value.converter.schemas.enable": "true",
        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false",
        "transforms.unwrap.delete.handling.mode": "rewrite"
    }
}

额外注意点

• 确保目标表字段类型与Kafka消息after中的字段类型完全一致，避免类型转换报错。
• 若之前出现“Could not find table”，检查table.name.format是否正确映射到目标表，确认表存在于指定的schema下。
• 删除操作处理：delete.handling.mode: rewrite会将删除转为带主键的null值upsert，若要直接删除数据，可改为drop（需主键配置正确）。

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