PostgreSQL Debezium Connector变更未在Kafka Topic显示及数据缺失问题

一、字段值为空/显示异常的核心原因 #

• 自定义类型name未被Debezium识别：firstname、lastname使用的PostgreSQL自定义类型name不在Debezium默认支持的标准SQL类型列表里，导致解析失败，字段值为空。
• date字段显示数字是默认行为：Debezium会把日期类型默认转成Epoch时间戳（毫秒数），并非异常，只是格式不符合预期。

二、后续变更未同步的关键诱因 #

• 逻辑复制权限/复制槽异常：Debezium依赖PostgreSQL逻辑复制捕获增量变更，若postgres用户没有membership schema的读权限、序列权限，或者逻辑复制槽未正确关联该表，都会导致增量变更无法被捕获。另外，postgresql.conf未开启逻辑复制（wal_level=logical等参数未配置）也会引发问题。
• 连接器配置缺失必要参数：未显式指定plugin.name=pgoutput（PostgreSQL 10+默认逻辑复制插件）、未配置类型映射，会导致连接器无法正确解析表结构和变更数据。
• 连接器初始化未扫描到新表：若先建表再注册连接器，可能连接器启动时未识别到新表，需要重新触发快照或重启连接器。

三、Topic未生成的常见情况 #

• 连接器启动失败：查看Debezium日志，大概率是权限不足、表不存在（注册连接器时membership.members还未创建）或配置错误导致连接器未正常启动，自然不会生成Topic。
• Topic命名误解：Debezium生成的Topic格式为${database.server.name}.${schema}.${table}，也就是dbserver1.membership.members，确认是否在Kafka UI中找错了名称。

1. 处理自定义类型name的解析
   在连接器配置中添加类型映射，将name类型映射为标准VARCHAR：

"database.type.mapping": "name=VARCHAR"

2. 调整日期字段显示格式
   添加配置将date类型转换为可读字符串：

"time.precision.mode": "connect",
"transforms": "unwrap,convertDate",
"transforms.convertDate.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.convertDate.field": "created",
"transforms.convertDate.target.type": "string",
"transforms.convertDate.format": "yyyy-MM-dd"

3. 确保逻辑复制权限与配置正常

• 给postgres用户授予必要权限：

ALTER USER postgres REPLICATION;
GRANT SELECT ON ALL TABLES IN SCHEMA membership TO postgres;
GRANT USAGE, SELECT ON SEQUENCES IN SCHEMA membership TO postgres;

• 检查postgresql.conf关键配置，确保开启逻辑复制：

wal_level = logical
max_wal_senders = 10
max_replication_slots = 10

• 若存在无效复制槽，清理后重新创建连接器：

-- 查看复制槽
SELECT * FROM pg_replication_slots;
-- 删除对应槽（替换为你的槽名）
SELECT pg_drop_replication_slot('dbserver1');

4. 更新连接器完整配置
   替换为以下配置重新注册连接器：

{
    "name": "membership-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "postgres",
        "database.port": "5432",
        "database.user": "postgres",
        "database.password": "postgres",
        "database.dbname": "postgres",
        "database.server.name": "dbserver1",
        "table.include.list": "membership.members",
        "plugin.name": "pgoutput",
        "database.type.mapping": "name=VARCHAR",
        "time.precision.mode": "connect",
        "transforms": "unwrap,convertDate",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "true",
        "transforms.convertDate.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.convertDate.field": "created",
        "transforms.convertDate.target.type": "string",
        "transforms.convertDate.format": "yyyy-MM-dd"
    }
}

5. 验证效果
   删除旧连接器，用新配置重新注册；检查连接器状态为RUNNING后，插入测试数据到membership.members，查看Kafka Topicdbserver1.membership.members中的消息，确认字段值完整、格式正确且增量变更能同步。

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