KafkaJDBCSinkConnector无法将消息发送到DLQ

在Kafka JDBC Sink Connector配置文件中设置error.policy参数为DLQ。在DLQ配置文件中设置存储和读取的位置，以便存储失败的消息并进行分析处理。以下是示例代码：

Kafka JDBC Sink Connector配置文件:

{ “name”: “jdbc-sink”, “connector.class”: “io.confluent.connect.jdbc.JdbcSinkConnector”, “tasks.max”: “1”, “topics”: “test-topic”, “connection.url”: “jdbc:postgresql://localhost/test_db”, “auto.create”: “true”, “insert.mode”: “upsert”, “key.converter”: “org.apache.kafka.connect.storage.StringConverter”, “value.converter”: “io.confluent.connect.avro.AvroConverter”, “value.converter.schema.registry.url”: “http://localhost:8081”, “errors.tolerance”: “all”, “errors.log.enable”: true, “errors.deadletterqueue.topic.name”: “dlq_test-topic”, “errors.deadletterqueue.context.headers.enable”: true, “errors.deadletterqueue.topic.replication.factor”: 1, “errors.deadletterqueue.topic.creat e” : true, “errors.deadletterqueue.kafka.bootstrap.servers”: “localhost:9092”, “errors.deadletterqueue.kafka.security.protocol”: “SASL_SSL”, “errors.deadletterqueue.kafka.ssl.truststore.location”: “/path/to/truststore” }

DLQ配置文件:

{ “name”: “dlq-jdbc-sink”, “connector.class”: “io.confluent.connect.jdbc.JdbcSinkConnector”, “tasks.max”: “1”, “topics”: “dlq_test-topic”, “connection.url”: “jdbc:postgresql://localhost/test_db”, “auto.create”: “true”, “insert.mode”: “insert”, “key.converter”: “org.apache.kafka.connect.storage.StringConverter”, “value.converter”: “io.confluent.connect.avro.AvroConverter”, “value.converter.schema.registry.url”: “http://localhost:8081” }