Debezium Kafka Connect复制槽延迟增加且失效，连接器仍运行的问题排查

我使用Confluent Kafka做数据流式处理，采用Postgres源连接器搭配JDBC sink连接器，源连接器配置如下：

{"name": "counterops_name",   
"config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "**.**.***.***",
    "database.port": 5432,
    "database.user": "***",
    "database.password": "****",
    "database.dbname": "***",
    "database.server.name": "**.**.***.***",
    "table.include.list": "schema.table",
    "snapshot.mode": "initial",
    "time.precision.mode": "connect",
    "database.history.kafka.bootstrap.servers": "broker1:9091,broker2:9091,broker3:9091,broker4:9091,broker5:9091",
    "database.history.kafka.topic": "schema-changes.schema",
    "topic.prefix": "prefix",
    "plugin.name": "pgoutput",
    "slot.name": "slot_name",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "https://broker3:8081,https://broker4:8081,https://broker5:8081",
    "key.converter.enhanced.avro.schema.support": true,
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schemas.enable": true,
    "value.converter.schema.registry.url": "https://broker3:8081,https://broker4:8081,https://broker5:8081",
    "transforms": "unwrap,removeFields",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.removeFields.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
    "transforms.removeFields.blacklist": "before,source",
    "include.schema.changes": true,
    "publication.name": "dbz_schema_table",
    "publication.autocreate.mode": "filtered",
    "tasks.max": "5",
    "topic.creation.default.replication.factor": 5,
    "topic.creation.default.partitions": 5,
    "poll.interval.ms": "1000",
    "offset.flush.interval.ms": "5000"
}}

运行几天后，复制槽变为inactive状态，数据停止流动，但连接器仍显示running。观察到复制状态从reserved变为extended，通过以下SQL查询到replicationSlotLag和confirmedLag数值已超过wal_size：

SELECT slot_name,
 pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as replicationSlotLag,
 pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) as confirmedLag,
 active
FROM pg_replication_slots;

请问：为何连接器处于运行状态时，replicationSlotLag和confirmedLag仍会持续增加？该如何解决这一延迟攀升问题？

可能的原因 #

• 连接器任务内部阻塞：连接器进程虽显示running，但内部任务可能因Kafka集群压力大、Schema Registry超时、网络波动等原因，停止消费WAL日志。PostgreSQL检测不到活跃的复制连接，标记复制槽为inactive，WAL日志持续堆积导致延迟上升。
• 任务数与资源不匹配：tasks.max=5，但如果目标Kafka主题分区数不足、或PostgreSQL表分区与任务数不匹配，会导致部分任务空转，真正处理数据的任务负载过高，跟不上WAL生成速度。
• WAL消费链路瓶颈：PostgreSQL侧数据写入量过大，加上Avro序列化慢、Schema Registry响应延迟、Kafka写入吞吐量不足等问题，导致连接器消费WAL的速度远低于生成速度。
• 复制槽会话异常：pgoutput插件的复制槽在连接器断开重连时，可能未正确恢复复制会话，PostgreSQL将槽标记为extended（无活跃连接），但连接器进程未检测到错误，仍显示running。
• 数据库历史主题阻塞：database.history.kafka.topic的消费如果卡住，会导致连接器无法处理Schema变更，进而阻塞整个WAL消费流程。

解决办法 #

• 排查任务日志并重启：查看Connect集群的任务日志，定位超时、连接失败等错误。若发现任务卡住，重启对应连接器任务或整个连接器。
• 匹配任务数与分区数：确保Kafka主题分区数不小于tasks.max，同时PostgreSQL表分区（若有）与任务数匹配，避免负载不均。当前主题分区为5，tasks.max=5配置合理，需确认每个任务都在处理数据。
• 优化消费与写入性能：
  • 微调poll.interval.ms：若数据量极大，可适当调小至500ms（避免过小给数据库带来额外压力）；
  • 优化Schema Registry：确保集群稳定、响应迅速，避免序列化阶段等待；
  • 配置Kafka生产参数：增大producer.batch.size、调整producer.linger.ms，提升Kafka写入吞吐量；
  • 增加连接器资源：给连接器分配更多CPU、内存，提升数据处理能力。
• 修复复制槽状态：
  • 先停止连接器，执行SELECT pg_drop_replication_slot('slot_name');删除无效复制槽，再重启连接器重新创建槽；
  • 检查PostgreSQL的max_replication_slots参数，确保有足够的槽资源可用。
• 配置监控告警：
  • 监控replicationSlotLag、confirmedLag、连接器任务状态、Kafka主题滞后量；
  • 设置阈值告警，及时发现并处理延迟问题，避免WAL堆积撑爆磁盘。

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