错误处理和从中断属性恢复

Overview

使用以下配置设置指定 MongoDB Kafka Source 连接器在遇到错误时的行为方式，并指定与恢复中断读取相关的设置。

设置

名称	说明
mongo.errors.tolerance	类型：string 说明：连接器遇到错误时是否继续处理消息。如果您希望连接器不再处理消息并报告遇到的问题，请将此项设置为 `"none"`。如果您希望连接器继续处理消息并忽略遇到的任何错误，请将此项设置为 `"all"`。重要：此属性覆盖 errors.tolerance 连接框架属性。默认：`"none"` 接受的值：`"none"` 或 `"all"`
mongo.errors.log.enable	类型：布尔值说明：连接器是否应在日志文件中报告错误。将此项设置为 `true`，以记录连接器遇到的所有错误日志。将此项设置为 `false` 以记录日志连接器不允许的错误。可以使用 `errors.tolerance` 或 `mongo.errors.tolerance` 设置指定连接器应允许哪些错误。重要：此属性覆盖 errors.log.enable 连接框架属性。默认：`false` 接受的值：`true` 或 `false`
mongo.errors.deadletterqueue.topic.name	Type: string Description: The name of topic to use as the dead letter queue. If you specify a value, the connector writes invalid messages to the dead letter queue topic as extended JSON strings. If you leave this setting blank, the connector does not write invalid messages to any topic. 重要：必须将 `errors.tolerance` 或 `mongo.errors.tolerance` 设置设置为 `"all"` 才能启用此属性。默认： `""` 接受的值：有效的 Kafka 主题名称
offset.partition.name	类型：string 说明：要使用的自定义偏移分区名称。当现有偏移包含无效的恢复令牌时，您可以使用此选项指示连接器启动新的变更流。如果将此设置留空，连接器会根据连接详细信息使用默认分区名称。要查看偏移分区命名策略，请参阅重置已存储偏移指南。默认： `""` 接受值：string。要了解有关命名分区的更多信息，请参阅 Apache Kafka API 文档中的 SourceRecord 部分。
heartbeat.interval.ms	类型：long 说明：连接器在发送心跳消息之间等待的毫秒数。当源记录未在指定时间间隔内发布时，连接器会发送心跳消息。此机制提高连接器针对低容量命名空间的可恢复性。心跳消息包含一个 `postBatchResumeToken` 数据字段。该字段的值包含连接器上次从变更流中读取的 MongoDB 服务器 oplog 条目。将此项设置为 `0`，禁用心跳消息。要了解更多信息，请参阅无效恢复令牌页面中的预防。默认：`0` 接受值：整数
heartbeat.topic.name	类型： string 说明：连接器应发布心跳消息的主题名称。您必须在 `heartbeat.interval.ms` 设置中提供正值才能启用此功能。默认：`__mongodb_heartbeats` 接受的值：有效的 Kafka 主题名称

使用单个消息转换的心跳

如果您在 Kafka Connect 部署中启用心跳并指定单一消息转换 (SMT) ，则必须从 SMT 中排除心跳消息。 SMT 是 Kafka 的一项功能，可让您对通过源 connector 传递的消息指定转换，而无需部署 Atlas Stream Processing 应用程序。

要从 SMT 中排除心跳消息，您必须创建谓词并将其应用于 SMT。谓词是 SMT 的一项功能，可让您在应用转换之前检查消息是否与条件语句匹配。

以下配置定义了IsHeartbeat谓词，用于匹配发送到默认心跳主题的心跳消息：

predicates=IsHeartbeat
predicates.IsHeartbeat.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsHeartbeat.pattern=__mongodb_heartbeats

以下配置使用前面的谓词从ExtractField转换中排除心跳消息：

transforms=Extract
transforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.Extract.field=<the field to extract from your Apache Kafka key>
transforms.Extract.predicate=IsHeartbeat
transforms.Extract.negate=true
# apply the default key schema as the extract transformation requires a struct object
output.format.key=schema

如果不从前面的转换中排除心跳消息，则connector在处理心跳消息后会引发以下错误：

ERROR WorkerSourceTask{id=mongo-source-0} Task threw an uncaught and unrecoverable exception. Task is being killed ...
...
Only Struct objects supported for [field extraction], found: java.lang.String

要学习；了解有关 SMT 的更多信息，请参阅如何通过 Confluence 在Kafka Connect 中使用单个消息转换。

要学习；了解有关谓词的更多信息，请参阅 Confluent 中的筛选器 (Apache Kafka)。

要学习；了解有关 ExtractField 转换的更多信息，请参阅 Confluence 中的 ExtractField。

要了解有关默认密钥模式的更多信息，请参阅默认模式页面。