Debezium 3.0按需触发PostgreSQL全量快照失败问题求助

问题背景 #

在Python项目中使用Debezium 3.0的PostgreSQL连接器，配置如下：

{
    "name": "dbz_name",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "host.docker.internal",
      "database.port": "5432",
      "database.user": "user",
      "database.password": "password",
      "database.dbname": "dbname",
      "database.server.name": "sname",
      "table.include.list": "schema.*",
      "plugin.name": "pgoutput",
      "publication.name": "dbz_publication",
      "slot.name": "dbz_slot",
      "key.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter.schemas.enable": false,
      "value.converter.schemas.enable": false,
      "topic.creation.default.replication.factor": 1,
      "topic.creation.default.partitions": 1,
      "topic.creation.default.cleanup.policy": "delete",
      "topic.prefix": "tprefix",
      "tombstones.on.delete": false,
      "snapshot.mode": "when_needed",

      "signals.data.collection": "signaling",
      "signal.actions": "execute-snapshot:execute-snapshot"
    }
  }

尝试通过以下Python代码触发指定表的增量快照，但信号消息已发送到signaling主题，目标表主题却未生成'r'类型快照消息：

from confluent_kafka import Producer
import json
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'signal-producer'
}

producer = Producer(conf)

topic = 'signaling'

signal_message = {"type": "execute-snapshot",
                  "data": {"data-collections": ["schema.table1"],
                           "type": "INCREMENTAL"}}

producer.produce(
    topic=topic,
    key="execute-snapshot",  
    value=json.dumps(signal_message)
)
producer.flush()

排查与修正步骤 #

1. 修正信号主题名称 #

Debezium的信号主题必须遵循{topic.prefix}.{连接器名称}.signals格式。你当前配置的topic.prefix为tprefix，连接器名称是dbz_name，所以正确的信号主题应该是tprefix.dbz_name.signals，而非直接使用signaling。

2. 修正信号消息的Key #

信号消息的Key必须设置为连接器名称（即dbz_name），Debezium通过这个Key识别目标连接器，错误的Key会导致连接器无法接收信号。

3. 确认PostgreSQL表的复制身份配置 #

对于INCREMENTAL类型快照，目标表必须开启REPLICA IDENTITY FULL，否则无法生成增量快照所需的变更数据。执行以下SQL确认并修改：

-- 查看当前配置
SELECT relreplident FROM pg_class WHERE relname = 'table1' AND relnamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'schema');

-- 如果不是'f'（FULL），执行修改
ALTER TABLE schema.table1 REPLICA IDENTITY FULL;

4. 检查连接器日志 #

查看Debezium连接器的运行日志，搜索signal或snapshot关键字，确认是否存在权限不足、信号格式错误或表匹配失败等报错。

修正后的Python代码示例 #

from confluent_kafka import Producer
import json
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'signal-producer'
}

producer = Producer(conf)

# 使用正确的信号主题名称
topic = 'tprefix.dbz_name.signals'

signal_message = {"type": "execute-snapshot",
                  "data": {"data-collections": ["schema.table1"],
                           "type": "INCREMENTAL"}}

# Key设置为连接器名称
producer.produce(
    topic=topic,
    key="dbz_name",  
    value=json.dumps(signal_message)
)
producer.flush()

内容的提问来源于stack exchange，提问作者Марина Лисниченко