连接器错误处理属性

Overview

使用以下配置设置指定 MongoDB Kafka connector 如何处理错误并配置死信队列（DLQ）。

有关按类别组织的接收器连接Connector配置设置列表，请参阅接收器连接器配置属性指南。

设置

名称	说明
mongo.errors.tolerance	类型： string 说明：连接器出错时是否继续处理消息。允许连接器覆盖 `errors.tolerance` Kafka 集群设置。当设置为 `none` 时，连接器会报告任何错误，并阻止对其余消息的进一步处理。当设置为 `all` 时，连接器会忽略任何有问题的消息。当设置为 `data` 时，连接器只能容忍数据错误，并对所有其他错误失败。要了解有关错误处理策略的更多信息，请参阅处理错误页面。此属性覆盖 errors.tolerance Connect 框架的属性。默认：从 `errors.tolerance` 设置继承值。可接受值： `"none"` 或 `"all"`
mongo.errors.log.enable	类型：布尔值说明：连接器是否应将错误详情（包括失败的操作）写入日志文件。连接器使用 `errors.tolerance` 或 `mongo.errors.tolerance` 设置将错误分类为“可容忍”或“不可容忍”。当设置为 `true` 时，连接器会记录“可容忍”和“不可容忍”错误。当设置为 `false` 时，连接器会记录“不可容忍”错误。此属性会覆盖 Connect Framework 的 errors.log.enable Connect 框架的属性。默认值： `false` 可接受值：`true` 或 `false`
errors.log.include.messages	类型：布尔值说明：连接器是否应在记录错误时包含无效消息。无效消息包含记录密钥、值和头部等数据。默认： `false` 接受值：`true` 或 `false`
errors.deadletterqueue.topic.name	Type: string Description: Name of topic to use as the dead letter queue. If blank, the connector does not send any invalid messages to the dead letter queue. To learn more about the dead letter queue, see the Dead Letter Queue Configuration Example. Default: `""` Accepted Values: A valid Kafka topic name
errors.deadletterqueue.context.headers.enable	Type: boolean Description: Whether the connector should include context headers when it writes messages to the dead letter queue. To learn more about the dead letter queue, see the Dead Letter Queue Configuration Example. To learn about the exceptions the connector defines and reports through context headers, see Bulk Write Exceptions. Default: `false` Accepted Values: `true` or `false`
errors.deadletterqueue.topic.replication.factor	Type: integer Description: The number of nodes on which to replicate the dead letter queue topic. If you are running a single-node Kafka cluster, you must set this to `1`. To learn more about the dead letter queue, see the Dead Letter Queue Configuration Example. Default: `3` Accepted Values: A valid number of nodes

批量写入异常

执行批量写入时，connector可以将以下异常作为上下文标头报告到死信队列（DLQ）：

名称

说明

WriteException

描述：
包含连接器遇到的 BulkWriteError 的详细信息。

消息格式：

此类按以下格式输出错误：

v=%d, code=%d, message=%s, details=%s

上述消息中的字段包含以下信息：

v： WriteException消息格式的版本。此字段有助于解析此异常生成的消息。对于版本 2.0 的connector ，消息格式的版本为 1。
code：与错误相关的代码。要学习；了解更多信息，请参阅 getCode() 方法文档。
message：与错误相关的消息。要学习；了解更多信息，请参阅 getMessage() 方法文档。
details：与 JSON 格式的错误相关的详细信息。要了解更多信息，请参阅以下方法文档：
- getDetails()
- toJson()

WriteConcernException

描述：
包含连接器遇到的 WriteConcernError 的详细信息。

消息格式：

此类按以下格式输出错误：

v=%d, code=%d, codeName=%d, message=%s, details=%s

上述消息中的字段包含以下信息：

v： WriteConcernException消息格式的版本。此字段有助于解析此异常生成的消息。对于版本 2.0 的connector ，消息格式的版本为 1。
code：与错误相关的代码。要学习；了解更多信息，请参阅 getCode() 方法文档。
codeName：与错误相关的代码名称。要学习；了解更多信息，请参阅 getCodeName() 方法文档。
message：与错误相关的消息。要学习；了解更多信息，请参阅 getMessage() 方法文档。
details：与 JSON 格式的错误相关的详细信息。要了解更多信息，请参阅以下方法文档：
- getDetails()
- toJson()

WriteSkippedException

描述：
通知 MongoDB 未尝试在以下场景中写入 SinkRecord：

connector向 MongoDB 发送有序批量写入操作
MongoDB 无法处理有序批量写入中的写入操作
MongoDB 不会尝试以有序批量写入的方式执行所有后续写入操作。

要了解如何将 Connector 设置为执行无序批量写入操作，请参阅《Connector 消息处理属性》页面。
消息格式：

此异常不产生任何消息。

要启用向死信队列（DLQ）报告批量写入异常，请使用以下connector配置：

errors.tolerance=all
errors.deadletterqueue.topic.name=<name of topic to use as dead letter queue>
errors.deadletterqueue.context.headers.enable=true

死信队列配置示例

Apache Kafka 2.6 版本增加了对处理错误记录的支持。 Kafka Connector会自动将其无法处理的消息发送到死信队列（DLQ）。一旦进入死信队列（DLQ），您可以检查错误记录，更新它们，然后重新提交以进行处理。

以下是启用死信队列（DLQ）主题example.deadletterqueue的示例配置。此配置指定死信队列（DLQ）和日志文件应记录无效消息，并且死信队列（DLQ）消息应包含上下文标头。

mongo.errors.tolerance=all
mongo.errors.log.enable=true
errors.log.include.messages=true
errors.deadletterqueue.topic.name=example.deadletterqueue
errors.deadletterqueue.context.headers.enable=true

要了解有关死信队列（DLQ）的更多信息，请参阅向主题写入错误和错误消息。