Kafka Connect Confluent JDBC Sink持久化延迟问题排查求助

搭建了PostgreSQL到PostgreSQL的CDC数据管道：

• 源端：PostgreSQL + Debezium Source Connector 2.1
• 目标端：PostgreSQL + Confluent JDBC Sink Connector 10.6.4（从Kafka读取数据）
• 当前问题：持续消息延迟，约40条消息滞后，消息发布速率约每秒5条

Confluent JDBC Sink Connector 调整的消费者配置 #

consumer.override.fetch.min.bytes=101200
consumer.override.fetch.max.wait.ms=1000
consumer.override.auto.commit.interval.ms=100
consumer.override.max.poll.records=5000
consumer.override.max.poll.interval.ms=300000
consumer.override.offset.flush.interval.ms=200

尝试Debezium Sink Connector 2.4.1时的配置及问题 #

配置：

consumer.override.offset.flush.timeout.ms=5000
consumer.override.max.poll.interval.ms=1000
consumer.override.enable.auto.commit=true
consumer.override.max.poll.records=1
consumer.override.auto.commit.interval.ms=10000

出现偏移量提交超时（Commit of offsets timed out），日志如下：

{"debug_level":"INFO","debug_timestamp":"2023-11-27 15:38:33,051","debug_thread":"task-thread-debezium_sink_connector3-0","debug_file":"WorkerSinkTask.java", "debug_line":"352","debug_message":"WorkerSinkTask{id=debezium_sink_connector3-0} Committing offsets asynchronously using sequence number 4: {mesh.public.glusr_usr-0=OffsetAndMetadata{offset=141102239, leaderEpoch=null, metadata=''}}"}
{"debug_level":"WARN","debug_timestamp":"2023-11-27 15:38:51,675","debug_thread":"task-thread-debezium_sink_connector3-0","debug_file":"WorkerSinkTask.java", "debug_line":"225","debug_message":"WorkerSinkTask{id=debezium_sink_connector3-0} Commit of offsets timed out"}
{"debug_level":"INFO","debug_timestamp":"2023-11-27 15:39:25,408","debug_thread":"task-thread-debezium_sink_connector3-0","debug_file":"WorkerSinkTask.java", "debug_line":"352","debug_message":"WorkerSinkTask{id=debezium_sink_connector3-0} Committing offsets asynchronously using sequence number 5: {mesh.public.glusr_usr-0=OffsetAndMetadata{offset=141103017, leaderEpoch=null, metadata=''}}"}
{"debug_level":"WARN","debug_timestamp":"2023-11-27 15:39:59,140","debug_thread":"task-thread-debezium_sink_connector3-0","debug_file":"WorkerSinkTask.java", "debug_line":"225","debug_message":"WorkerSinkTask{id=debezium_sink_connector3-0} Commit of offsets timed out"} 

• 目标端单条upsert查询仅耗时1ms，但Sink端仍有约10秒持久化延迟
• 尝试多种Kafka消费者配置未获理想效果
• 配置变更影响消费者延迟，但无法明确关联关系

一、Kafka集群与Connector任务资源排查 #

1. Kafka Broker性能检查：查看Kafka集群的磁盘IO、CPU、内存使用率，重点关注目标Topic的分区Leader所在Broker是否有资源瓶颈。如果Broker磁盘写入慢，会直接拖慢偏移提交速度，进而影响Sink的消息处理节奏。
2. Connector任务资源优化：确保Sink Connector的任务数和Topic分区数匹配（建议任务数=分区数），避免单任务扛过多分区导致过载。同时检查运行Connector的Worker节点CPU、内存是否充足，查看Worker的JVM日志，确认是否存在频繁GC的情况。

二、Confluent JDBC Sink Connector优化 #

1. 开启批量写入提升吞吐量：单条upsert快不代表批量效率高，调整以下配置：
   • batch.size：设置为100-500（默认100），根据实际情况调高，减少数据库交互次数
   • 确认insert.mode=upsert且配置了pk.fields指定主键，避免无主键导致的全表扫描
   • auto.create=false：关闭自动建表，减少Connector频繁检查表结构的额外开销
   • connection.max.idle.ms：调整数据库连接池的空闲超时时间，避免频繁创建销毁连接
2. 消费者配置回退与微调：当前部分配置可能反而增加延迟，建议先恢复默认再逐步调整：
   • 把fetch.min.bytes改回默认1，fetch.max.wait.ms改回默认500，避免消费者为凑够字节数等待过长时间
   • max.poll.records=5000过大，单批次处理5000条可能导致处理超时，建议调低到1000以内，配合批量写入配置
   • auto.commit.interval.ms=100过于频繁，调整为1000-5000，减少偏移提交的开销

三、Debezium Sink Connector超时问题修复 #

之前的Debezium Sink配置存在明显不合理，调整如下：

• max.poll.interval.ms=1000过小，单条消息处理超过1秒就会触发超时，改成300000（5分钟），和Confluent Sink配置保持一致
• max.poll.records=1会导致单批次仅处理1条消息，吞吐量极低，调高到100-500
• offset.flush.timeout.ms=5000适当调高到10000，给偏移提交足够的时间
• 关闭enable.auto.commit，让Connector自行管理偏移提交（Debezium Sink默认是手动提交，自动提交容易出问题）

四、端到端延迟定位 #

1. 在Debezium Source的消息中通过transforms添加源端时间戳字段，Sink端写入目标库时记录写入时间，计算每条消息的端到端延迟，明确延迟发生在Kafka拉取、Sink处理还是数据库写入阶段。
2. 用Kafka自带工具监控消费滞后：

   kafka-consumer-groups.sh --describe --group <sink-connector-group-id> --bootstrap-server <kafka-brokers>
   

   查看CURRENT-OFFSET和LOG-END-OFFSET的差值，确认滞后是持续增加还是稳定在40条。

五、数据库层面补充检查 #

1. 检查目标端PostgreSQL的max_connections是否足够，Connector的连接池能否获取到足够连接。
2. 查看目标表的锁情况和执行计划：
   • 查询pg_locks视图，确认是否有长时间持有的锁
   • 通过pg_stat_statements查看upsert语句的实际执行计划，排查是否存在隐式全表扫描或索引失效的情况

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