Kafka JDBC Sink Connector异常消息无法进入DLQ问题求助

问题描述 #

搭建Kafka JDBC Sink Connector用于从original_topic读取消息写入MySQL表，当部分消息违反外键约束（如地址表写入不存在的用户ID）时，连接器直接崩溃终止，无法将错误消息转入Dead Letter Queue（DLQ）以继续处理后续消息，配置的errors.tolerance=all未生效。

连接器配置 #

{
    "name": "connector_name",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "errors.log.include.messages": "true",
        "connection.password": "Password",
        "tasks.max": "1",
        "transforms": "unwrap",
        "max.retries": "0",
        "retry.backoff.ms": "5000",
        "errors.deadletterqueue.context.headers.enable": "true",
        "auto.evolve": "true",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "errors.log.enable": "true",
        "insert.mode": "upsert",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "max.poll.records": "1",
        "topics": "original_topic",
        "batch.size": "1",
        "key.converter.schemas.enable": "true",
        "connection.user": "Username",
        "errors.deadletterqueue.topic.name": "dlq_topic",
        "name": "connector_name",
        "value.converter.schemas.enable": "true",
        "errors.tolerance": "all",
        "connection.url": "url",
        "pk.fields": "id",
        "pk.mode": "record_key"
    }
}

异常信息 #

ERROR WorkerSinkTask{id=sink-v3-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. (org.apache.kafka.connect.runtime.WorkerSinkTask:559)

org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: Exception chain:

java.sql.BatchUpdateException: Cannot add or update a child row: a foreign key constraint fails

java.sql.SQLIntegrityConstraintViolationException: Cannot add or update a child row: a foreign key constraint fails

[2023-06-20 19:33:07,386] DEBUG WorkerSinkTask{id=sink-v3-0} Skipping offset commit, no change since last commit (org.apache.kafka.connect.runtime.WorkerSinkTask:427)

[2023-06-20 19:33:07,386] DEBUG WorkerSinkTask{id=sink-v3-0} Finished offset commit successfully in 0 ms for sequence number 1: null (org.apache.kafka.connect.runtime.WorkerSinkTask:264)

[2023-06-20 19:33:07,386] ERROR WorkerSinkTask{id=sink-v3-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:179)

...

at io.confluent.connect.jdbc.sink.JdbcSinkTask.getAllMessagesException(JdbcSinkTask.java:190)

at io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:133)

排查与解决方案 #

核心原因分析 #

从异常栈可见，连接器抛出getAllMessagesException，说明即使配置了batch.size=1，JDBC Sink仍将单条消息的失败判定为全批次失败，触发不可恢复错误导致任务终止，而非将单条消息路由到DLQ。这是因为upsert模式下，连接器的批量处理逻辑会将单条失败消息升级为批次级异常，绕过了DLQ的单个消息错误处理逻辑。

具体解决步骤 #

1. 确认DLQ Topic的有效性

   • 手动创建DLQ Topic（若未自动创建）：

     kafka-topics --create --topic dlq_topic --bootstrap-server <你的Bootstrap地址> --partitions 1 --replication-factor 1
     

   • 确保连接器对DLQ Topic有读写权限。

2. 补充DLQ配置项
   在连接器配置中添加：

   "errors.deadletterqueue.topic.replication.factor": "1"
   

   （根据集群实际情况调整副本数，确保Topic能正常创建）

3. 检查全局配置覆盖问题
   查看Kafka Connect Worker的全局配置文件，确认未设置全局errors.tolerance=none（该配置会覆盖连接器级别的errors.tolerance=all）。

4. 验证连接器版本兼容性
   确保使用的Confluent JDBC Sink Connector版本≥5.4.0，DLQ功能在该版本后才稳定支持异常消息路由。

5. 调整批量处理与错误逻辑

   • 保持batch.size=1和max.poll.records=1的配置，确保每次仅处理单条消息。
   • 若upsert模式仍触发批次异常，可临时切换为insert.mode=insert测试DLQ是否生效，排除upsert逻辑的影响。

6. 前置过滤错误消息
   添加Debezium或自定义Transform，在消息进入连接器前验证外键关联的记录是否存在，提前过滤无效消息：

   "transforms": "unwrap,filterInvalid",
   "transforms.filterInvalid.type": "org.apache.kafka.connect.transforms.Filter$Value",
   "transforms.filterInvalid.condition": "$.user_id is not null and exists(select id from users where id = $.user_id)"
   

   （需结合实际消息结构调整过滤条件）

7. MySQL端优化（可选）
   若业务允许，可修改MySQL外键约束为ON DELETE SET NULL或ON UPDATE CASCADE，避免硬失败，但需评估业务影响。

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