我正在使用jdbc接收器连接器，主题中有一条错误消息。我知道为什么消息是坏的（它是失败的，因为fk约束冲突，因为与生产者的问题）。辅助任务报告的错误是：
org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:323)
org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:226)
org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:194)
org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
java.util.concurrent.FutureTask.run(FutureTask.java:266)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)\nCaused by: org.apache.kafka.connect.errors.ConnectException: java.sql.SQLException: java.sql.BatchUpdateException: Cannot add or update a child row: a foreign key constraint fails (`sensorium`.`reading`, CONSTRAINT `reading_ibfk_1` FOREIGN KEY (`sensorId`) REFERENCES `sensor` (`id`))\ncom.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException: Cannot add or update a child row: a foreign key constraint fails (`sensorium`.`reading`, CONSTRAINT `reading_ibfk_1` FOREIGN KEY (`sensorId`) REFERENCES `sensor` (`id`))\n
io.confluent.connect.jdbc.sink.JdbcSinkTask.put(JdbcSinkTask.java:86)
org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:565)\n\t... 10 more\nCaused by: java.sql.SQLException: java.sql.BatchUpdateException:
Cannot add or update a child row: a foreign key constraint fails
(`sensorium`.`reading`, CONSTRAINT `reading_ibfk_1` FOREIGN KEY (`sensorId`) REFERENCES `sensor`
(`id`))\ncom.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolation
Exception: Cannot add or update a child row: a foreign key constraint
fails (`sensorium`.`reading`, CONSTRAINT `reading_ibfk_1` FOREIGN KEY
(`sensorId`) REFERENCES `sensor` (`id`))我想发生的是跳过这个坏消息。所以我试着 "errors.tolerance": "all" . 接收器连接器的完整配置如下所示：
{
"name": "reading-sink2",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": 4,
"topics": "READING_MYSQL",
"key.converter.schema.registry.url": "http://localhost:8081",
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter.schema.registry.url": "http://localhost:8081",
"connection.url": "jdbc:mysql://localhost:3306/sensorium?user=app&password=tQpRMCzHlAeu6kQIBk4U",
"auto.create": true,
"table.name.format": "reading",
"errors.tolerance": "all"
}
}但是同样的错误正在被记录，消息不会被跳过，后续的消息也不会被处理。
为什么是 errors.tolerance: all 不按预期工作？
2条答案
按热度按时间gxwragnw1#
您可以使用
kafka-consumer-groups工具：更多信息请参见此处。
我已经记录了Flume的改进建议，请随意投票：https://github.com/confluentinc/kafka-connect-jdbc/issues/721
bwntbbo32#
errors.tolerance属性是指在转换（与kafka connect schema的消息转换）或转换消息（应用单个消息转换）期间发生的错误。您不能跳过/接受异常，这些异常是在
SinkTask::put(Collection<SinkRecord> records)或者SourceTask::poll()在您的情况下，会抛出异常SinkTask::put(...)io.confluent.connect.jdbc.sink.jdbcsinktask.put（jdbcsinktask。java:86)关于类似问题的问题：
kafka连接接收器任务忽略容差限制
kafka connect-jdbc接收器sql异常
apache kafka jdbc连接器-serializationexception:未知的魔法字节
您可以在以下博客的confluent页面上了解更多信息：https://www.confluent.io/blog/kafka-connect-deep-dive-error-handling-dead-letter-queues