Debezium PostgreSQL连接器不识别table.include.list，无法捕获指定表变更

环境信息 #

• Debezium Embedded版本：2.4.1.Final
• 使用连接器：PostgreSQL连接器

当前配置代码 #

io.debezium.config.Configuration.create()
                .with("name", "author-connector")
                .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "/tmp/offsets.dat")
                .with("offset.flush.interval.ms", "60000")
                .with("database.hostname", "localhost")
                .with("database.port", "5432")
                .with("database.user", postgres)
                .with("database.password", postgres)
                .with("database.dbname", postgres)
                .with("database.include.list", postgres)
                .with("schema.include.list", "admin")
                .with("table.include.list", "admin.connector")
                .with("include.schema.changes", "true")
                .with("database.server.name", "author-server")
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                .with("database.history.file.filename", "/tmp/dbhistory.dat")
                .with("plugin.name", "pgoutput")
                .with("signal.enabled.channels", "source")
                .with("topic.prefix", "admin")
                .with("quarkus.http.port", "7001")
                .with("signal.data.collection", "admin.signal_table")
                .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                .with("offset.storage.file.filename", "/tmp/offsets.dat")
                .with("offset.flush.interval.ms", "60000")
                .build();

出现的问题 #

1. 连接器与数据库连接正常，但admin.connector表的INSERT/UPDATE/DELETE变更完全无法被捕获
2. 尝试通过信号表执行增量快照时，触发如下错误：

ractIncrementalSnapshotChangeEventSource : Schema not found for table 'admin.connector', known tables [admin.signal_table]

排查与解决方法 #

1. 清理重复配置 #

配置中重复定义了offset.storage、offset.storage.file.filename、offset.flush.interval.ms三项，重复配置可能导致加载逻辑异常，删除重复项，保留一份即可。

2. 验证PostgreSQL逻辑复制配置 #

使用pgoutput插件必须满足以下条件：

• 数据库用户需拥有REPLICATION权限，同时对admin.connector表有SELECT权限，执行以下SQL确认：

  SELECT rolreplication FROM pg_roles WHERE rolname = 'postgres';
  SELECT has_table_privilege('postgres', 'admin.connector', 'SELECT');
  

• 修改postgresql.conf开启逻辑复制：

  wal_level = logical
  max_wal_senders = 10
  max_replication_slots = 10
  

  修改后重启PostgreSQL服务，再检查复制槽是否存在：

  SELECT * FROM pg_replication_slots WHERE slot_name = 'author-connector';
  

3. 刷新表元数据 #

连接器启动时未加载到admin.connector的元数据，导致无法捕获变更：

• 删除/tmp/dbhistory.dat和/tmp/offsets.dat文件，重启连接器触发全量元数据扫描
• 如果admin.connector表是在连接器启动后创建的，可通过信号表发送ADJUST_SCHEMA信号刷新元数据：

  INSERT INTO admin.signal_table (id, type, data)
  VALUES ('schema-refresh', 'ADJUST_SCHEMA', '{"source": {"server": "author-server"}}');
  

4. 检查增量快照信号格式 #

执行增量快照的信号记录格式必须正确，确保插入的JSON数据中指定的表名与实际一致：

INSERT INTO admin.signal_table (id, type, data)
VALUES ('snapshot-1', 'execute-snapshot', '{"data-collections": ["admin.connector"], "type": "INCREMENTAL"}');

同时确认信号表结构正确，需包含id(VARCHAR)、type(VARCHAR)、data(JSONB)三个字段。

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