Debezium重启后未读取Kafka偏移量主题最新消息求助

问题根源 #

你直接向偏移量主题发送消息但未生效，核心原因有两个：

1. Debezium通过专属消费者组读取偏移量主题，你的消费者组偏移仍停留在0位置，重启后会从第一条消息开始读取；
2. 你发送的偏移量消息格式不符合Debezium规范，导致解析失败触发偏移重置。

以下是两种可行的调整方法：

方法一：用Kafka官方脚本重置消费者组偏移 #

1. 定位Debezium消费者组 #

执行命令列出所有Kafka消费者组，找到Debezium对应的组（通常格式为debezium-<连接器名称>）：

kafka-consumer-groups.sh --bootstrap-server <Kafka Broker地址:端口> --list

2. 查看当前偏移状态 #

确认目标消费者组在偏移量主题（格式为__debezium-offset-<连接器名称>）上的偏移：

kafka-consumer-groups.sh --bootstrap-server <Kafka Broker地址:端口> --describe --group <Debezium消费者组名称>

3. 重置偏移到最新位置 #

将消费者组在偏移量主题上的偏移强制设置为最新位置：

kafka-consumer-groups.sh --bootstrap-server <Kafka Broker地址:端口> --reset-offsets --to-latest --topic __debezium-offset-<连接器名称> --group <Debezium消费者组名称> --execute

执行完成后重启Debezium连接器，即可读取偏移量主题的最后一条有效消息。

方法二：用kcat发送合规偏移量消息并调整偏移 #

1. 构造符合规范的偏移量消息 #

MySQL场景下的Debezium偏移量JSON格式示例（替换为你的实际binlog信息）：

{
  "source": {
    "server_id": 1,
    "ts_sec": 1690000000,
    "gtid": null,
    "file": "mysql-bin.000005",
    "pos": 12345,
    "row": 0,
    "snapshot": false,
    "thread": null,
    "db": "你的数据库名",
    "table": null
  },
  "transaction_id": null,
  "ts_ms": 1690000000000,
  "file": "mysql-bin.000005",
  "pos": 12345
}

2. 用kcat发送消息 #

注意消息key必须与Debezium原有格式一致（通常为<连接器名称>:<MySQL服务器名称>）：

kcat -P -b <Kafka Broker地址:端口> -t __debezium-offset-<连接器名称> -p 0 -k "<连接器名称>:<MySQL服务器名称>" <<EOF
{"source": {"server_id": 1, "ts_sec": 1690000000, "gtid": null, "file": "mysql-bin.000005", "pos": 12345, "row": 0, "snapshot": false, "thread": null, "db": "你的数据库名", "table": null}, "transaction_id": null, "ts_ms": 1690000000000, "file": "mysql-bin.000005", "pos": 12345}
EOF

3. 重置消费者组偏移 #

重复方法一中的步骤3，将消费者组偏移设置为最新位置，确保Debezium重启后读取这条新消息。

额外注意事项 #

• 操作前请暂停Debezium连接器，避免并行读写偏移量主题；
• 偏移量主题多分区场景下，需确认操作的是连接器对应的分区；
• 禁止发送格式错误的消息，否则会触发Debezium自动重置偏移到初始位置。

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