JdbcSinkConnector延迟重试配置异常：5分钟重试未生效排查

问题场景 #

已将connection.backoff.ms、retry.backoff.ms参数设置为5分钟（300000ms），期望Kafka向断开连接的数据库发送消息失败后，重试间隔为5分钟。但重启数据库后，消息立即送达，并未等待5分钟，询问是否存在其他影响该逻辑的参数。

使用的是Confluent Kafka转数据库Demo环境，连接器配置如下：

curl -X PUT http://localhost:8083/connectors/jdbc_postgres/config \
     -H "Content-Type: application/json" -d '{
"connector.class":"io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url":"jdbc:postgresql://postgres:5432/postgres",
"topics":"test01",
"key.converter":"org.apache.kafka.connect.storage.StringConverter",
"value.converter":"io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url":"http://schema-registry:8081",
"connection.user":"postgres",
"connection.password":"postgres",
"auto.create":true,
"auto.evolve":true,
"insert.mode":"upsert",
"pk.mode":"record_key",
"pk.fields":"MESSAGE_KEY",
"transforms":"Filter",
"transforms.Filter.type":"org.apache.kafka.connect.transforms.Filter",
"transforms.Filter.predicate":"DropNull",
"predicates":"DropNull",
"predicates.DropNull.type":"org.apache.kafka.connect.transforms.predicates.RecordIsTombstone",
"errors.log.enable":"true",
"errors.tolerance":"all",
"errors.deadletterqueue.topic.name":"dead_test01",
"errors.deadletterqueue.topic.replication.factor":"-1",
"connection.backoff.ms":"300000"
}'

关键影响参数及逻辑说明 #

1. connection.backoff.ms与retry.backoff.ms的区别

   • connection.backoff.ms：仅在数据库连接完全断开时生效，控制连接器尝试重新建立连接的间隔。一旦数据库恢复、连接重建成功，连接器会立即处理积压消息，不会再等待该间隔。
   • retry.backoff.ms：针对数据库连接正常但单个消息发送失败的场景（如主键冲突、字段类型不匹配），控制重试间隔。你的配置中未显式设置该参数，需确认是否实际配置生效。

2. errors.tolerance
   设置为all时，连接器遇到错误不会停止，会继续处理后续消息，但该参数不直接控制重试间隔，仅决定错误容忍策略。若配合errors.retry.timeout.ms（默认无超时限制），会限制重试的总时长，超时后消息将被转发到死信队列。

3. max.retries
   默认值为10，控制单个消息的最大重试次数。若未修改，当消息发送失败次数达到阈值后，会被转入死信队列（若已配置），而非无限重试。

问题原因总结 #

你遇到的情况是：数据库断开后，连接器按connection.backoff.ms间隔尝试重连；当数据库重启、连接器成功建立新连接后，会立即处理积压的未发送消息——这是连接器的默认行为，目的是尽快恢复数据同步，而非等待重试间隔。原生JDBC Sink连接器没有“连接恢复后延迟发送消息”的配置，若需此逻辑，需通过自定义拦截器或外部机制实现。

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