所有源连接器配置属性

connection.uri

必需

Type: string

Description:
The URI connection string to connect to your MongoDB instance or cluster.

To learn more, see Connect to MongoDB.

重要提示：为避免在 connection.uri 设置中暴露身份验证凭证，请使用ConfigProvider并设立适当的配置参数。

默认：mongodb://localhost:27017,localhost:27018,localhost:27019
接受值：MongoDB URI 连接字符串

database

类型：string

说明：
要监视更改的数据库名称。如果未设置，连接器会监控所有数据库的更改。

默认：""
已接受值：单个数据库名称

集合

类型： string

说明：
数据库中集合的名称，用于监视变更。如果未设置，连接器会监控所有集合的更改。

重要： 如果您的 database 配置设置为 ""，连接器将忽略 collection 设置。

默认：""
已接受值：单个集合名称

server.api.version

Type: string

Description:
The Stable API version you want to use with your MongoDB cluster. For more information on the Stable API and versions of MongoDB server that support it, see the Stable API guide.

Default: ""
Accepted Values: An empty string or a valid Stable API version.

server.api.deprecationErrors

Type: boolean

Description:
When set to true, if the connector calls a command on your MongoDB instance that's deprecated in the declared Stable API version, it raises an exception.

You can set the API version with the server.api.version configuration option. For more information on the Stable API, see the MongoDB manual entry on the Stable API.

Default: false
Accepted Values: true or false

server.api.strict

Type: boolean

Description:
When set to true, if the connector calls a command on your MongoDB instance that's not covered in the declared Stable API version, it raises an exception.

You can set the API version with the server.api.version configuration option. For more information on the Stable API, see the MongoDB manual entry on the Stable API.

Default: false
Accepted Values: true or false

topic.prefix

类型：string

说明：
指定连接器将变更流事件发布到的目标 Kafka 主题名称的第一部分。目标主题名称的组成：topic.prefix 值，后跟数据库和集合名称，由 topic.separator 属性中指定的值分隔。

要了解更多信息，请参阅主题命名规则中的示例。

默认：""
接受的值：由 ASCII 字母数字字符组成的字符串，包括“.”、“-”和“_”

topic.suffix

类型：string

说明：
指定连接器将变更流事件发布到的目标 Kafka 主题名称的最后一部分。目标主题名称由数据库和集合名称后跟 topic.suffix 值组成，并由 topic.separator 属性中指定的值分隔。

要了解更多信息，请参阅主题命名后缀中的示例。

默认：""
接受的值：由 ASCII 字母数字字符组成的字符串，包括“.”、“-”和“_”

topic.namespace.map

Type: string

Description:
Specifies a JSON mapping between change stream document namespaces and topic names.

You can use to topic.namespace.map property to specify complex mappings. This property supports regex and wildcard matching.

To learn more about these behaviors and view examples, see Topic Namespace Map.

Default: ""
Accepted Values: A valid JSON object

topic.separator

类型：string

说明：
指定连接器用来连接用于创建主题名称的值的字符串。连接器将记录发布到主题，其名称是通过按以下顺序连接以下字段的值形成的：

1. topic.prefix

2. database

3. collection

4. topic.suffix

例如，以下配置指示连接器将变更流文档从 db 数据库的 coll 集合发布到 prefix-db-coll 主题：

topic.prefix=prefix
database=db
collection=coll
topic.separator=-

IMPORTANT: When you use the topic.separator property, note that it doesn't affect how you define the topic.namespace.map property. The topic.namespace.map property uses MongoDB namespaces which you must always specify with a . character to separate the database and collection name.

Default: "."
Accepted Values: A string

topic.mapper

类型：string

说明：
定义自定义主题映射逻辑的 Java 类。

默认：com.mongodb.kafka.connect.source.topic.mapping.DefaultTopicMapper
接受值：TopicMapper 类实现的有效完整类名。

管道

类型：string

说明：
要在变更流中运行的聚合管道数组。您必须为变更流事件文档而不是 fullDocument 字段配置此设置。

示例：

[{"$match": { "$and": [{"operationType": "insert"}, {"fullDocument.eventId": 1404 }] } }]

有关更多示例，请参阅:

• 自定义管道以筛选变更事件

• 监听多个来源的变更

默认："[]"
接受值：有效的聚合管道阶段

change.stream.full.document

Type: string

Description:
Determines what values your change stream returns on update operations.

The default setting returns the differences between the original document and the updated document.

The updateLookup setting returns the differences between the original document and updated document as well as a copy of the entire updated document at a point in time after the update.

The whenAvailable setting returns the updated document, if available.

The required setting returns the updated document and raises an error if it is not available.

For more information on how this change stream option works, see Lookup Full Document for Update Operations in the MongoDB manual.

Default: ""
Accepted Values: "", "updateLookup", "whenAvailable", or "required"

change.stream.show.expanded.events

Type: boolean

Description:
Determines if change streams notifies for DDL events such as
createIndexes and
dropIndexes
events. To learn more, see showExpandedEvents.

This setting is required to show
updateDescription.disambiguatedPaths in update events, which
clarifies changes that involve ambiguous fields. To learn more,
see disambiguatedPaths.

Default: false
Accepted Values: true or false

change.stream.full.document.before.change

Type: string

Description:
Configures the document pre-image your change stream returns on update operations. The pre-image is not available for source records published while copying existing data, and the pre-image configuration has no effect on copying.

To learn how to configure a collection to enable pre-images, see Change Streams with Document Pre- and Post-Images in the MongoDB manual.

The default setting suppresses the document pre-image.

The whenAvailable setting returns the document pre-image if it's available, before it was replaced, updated, or deleted.

The required setting returns the document pre-image and raises an error if it is not available.

Default: ""
Accepted Values: "" or "whenAvailable" or "required"

publish.full.document.only

Type: boolean

Description:
Whether to return only the fullDocument field from the change stream event document produced by any update event. The fullDocument field contains the most current version of the updated document. To learn more about the fullDocument field, see the update Event in the Server manual.

When set to true, the connector overrides the change.stream.full.document setting and sets it to updateLookup so that the fullDocument field contains updated documents.

Default: false
Accepted Values: true or false

publish.full.document.only.tombstone.on.delete

类型：布尔值

说明：
删除文档时是否返回逻辑删除事件。逻辑删除事件包含具有 null 值的已删除文档的密钥。此设置仅在 publish.full.document.only 为 true 时应用。

默认: false
接受的值: true 或 false

change.stream.document.key.as.key

类型： 布尔值

说明：
如果存在文档密钥，是否使用文档密钥作为源记录密钥。

当设置为 true 时，连接器会将已删除文档的密钥添加到 tombstone 事件中。当设置为 false 时，连接器会使用恢复令牌作为 tombstone 事件的源密钥。

默认: true
允许的值: true 或 false

排序规则

Type: string

Description:
A JSON collation document that specifies language-specific ordering rules that MongoDB applies to the documents returned by the change stream.

Default: ""
Accepted Values: A valid collation JSON document

batch.size

类型： int

说明：
变更流游标批处理大小。

默认: 0
接受值: 整数

poll.await.time.ms

类型： long

说明：
在返回空批处理之前，服务器等待新数据更改以报告给变更流游标的最长时间（以毫秒为单位）。

默认: 5000
接受值: 整数

poll.max.batch.size

类型： int

说明：
轮询变更流游标以获取新数据时，在单个批处理中读取的最大文档数。您可以使用此设置来限制连接器内部缓冲的数据量。

默认: 1000
接受值: 整数

output.format.key

类型：string

说明：
指定源连接器输出键文档的数据格式。

默认：json
接受的值：bson、json、schema

output.format.value

类型：string

说明：
指定源连接器将值文档输出为哪种数据格式。

连接器支持 Protobuf 作为输出数据格式。您可以通过指定 schema 值并安装和配置 Kafka Connect Protobuf 转换器来启用此格式。

默认：json
接受的值：bson、json、schema

output.json.formatter

类型：string

说明：
连接器用于输出数据的 JSON 格式化程序的类名称。

默认：

com.mongodb.kafka.connect.source.json.formatter.DefaultJson

可接受值：
您的自定义 JSON 格式化程序完整类名或以下内置格式化程序类名之一：

com.mongodb.kafka.connect.source.json.formatter.DefaultJson
com.mongodb.kafka.connect.source.json.formatter.ExtendedJson
com.mongodb.kafka.connect.source.json.formatter.SimplifiedJson

要了解有关这些输出格式的更多信息，请参阅 JSON 格式化程序。

output.schema.key

类型：string

说明：
为 SourceRecord 的键文档指定 Avro 模式定义。

要了解关于 Avro 模式的更多信息，请参阅 数据格式指南 中的 Avro 。

默认：

{
  "type": "record",
  "name": "keySchema",
  "fields" : [ { "name": "_id", "type": "string" } ]"
}

接受值：有效的 AVRO 模式

output.schema.value

类型：string

说明：
为 SourceRecord 的值文档指定 AVRO 模式定义。

要了解关于 Avro 模式的更多信息，请参阅 数据格式指南 中的 Avro 。

默认：

{
  "name": "ChangeStream",
  "type": "record",
  "fields": [
    { "name": "_id", "type": "string" },
    { "name": "operationType", "type": ["string", "null"] },
    { "name": "fullDocument", "type": ["string", "null"] },
    { "name": "ns",
      "type": [{"name": "ns", "type": "record", "fields": [
                {"name": "db", "type": "string"},
                {"name": "coll", "type": ["string", "null"] } ]
               }, "null" ] },
    { "name": "to",
      "type": [{"name": "to", "type": "record",  "fields": [
                {"name": "db", "type": "string"},
                {"name": "coll", "type": ["string", "null"] } ]
               }, "null" ] },
    { "name": "documentKey", "type": ["string", "null"] },
    { "name": "updateDescription",
      "type": [{"name": "updateDescription",  "type": "record", "fields": [
                 {"name": "updatedFields", "type": ["string", "null"]},
                 {"name": "removedFields",
                  "type": [{"type": "array", "items": "string"}, "null"]
                  }] }, "null"] },
    { "name": "clusterTime", "type": ["string", "null"] },
    { "name": "txnNumber", "type": ["long", "null"]},
    { "name": "lsid", "type": [{"name": "lsid", "type": "record",
               "fields": [ {"name": "id", "type": "string"},
                             {"name": "uid", "type": "string"}] }, "null"] }
  ]
}

接受值：有效的 JSON schema

output.schema.infer.value

类型：布尔值

说明：
连接器是否应推理出 SourceRecord 值文档的模式。由于连接器以隔离性处理每个文档，因此连接器可能会生成许多模式。

重要：仅当您将 output.format.value 设置设置为 schema 时，连接器才会读取此设置。

默认：false
接受的值：true 或 false

startup.mode

类型： string

说明：
指定当没有可用的源偏移时，连接器应如何启动。恢复变更流需要恢复令牌，连接器将从源偏移获取该令牌。如果没有可用的源偏移，则连接器可能会忽略全部或部分现有源数据，或者可能会先复制所有现有源数据，然后继续处理新数据。

如果为 startup.mode=latest，连接器会忽略所有现有源数据。

如果为 startup.mode=timestamp，连接器将激活 startup.mode.timestamp.* 属性。如果未配置属性，timestamp 等同于 latest 。

如果 startup.mode=copy_existing，连接器会将所有现有源数据复制到变更流事件。此设置相当于已弃用的设置 copy.existing=true。

如果任何系统在源Connector从数据库转换现有数据时更改数据库中的数据， MongoDB可能会生成重复的变更流事件以反映最新更改。由于数据复制所依赖的变更流事件是幂等的，因此复制的数据最终是一致的，符合“至少一次”交付保证。

默认：latest
接受的值：latest、timestamp、copy_existing

startup.mode.timestamp.start.at.operation.time

Type: string

Description:
Actuated only if startup.mode=timestamp. Specifies the starting point for the change stream.

To learn more about Change Stream parameters, see $changeStream (aggregation) in the MongoDB manual.

Default: ""
Accepted Values:

• 自纪元以来的整数秒数（十进制格式）（例如，30）

• ISO-8601 格式的瞬间，精度为一秒（例如，1970-01-01T00:00:30Z）

• 规范扩展 JSON (v2) 格式的 BSON 时间戳（例如 {"$timestamp": {"t": 30, "i": 0}}）

startup.mode.copy.existing.namespace.regex

类型： string

说明：
连接器用于匹配要从中复制数据的命名空间的正则表达式。命名空间描述由句点分隔的 MongoDB 数据库名称和集合（示例，databaseName.collectionName）。

示例，以下正则表达式设置与 stats 数据库中以“page”开头的集合相匹配：

startup.mode.copy.existing.namespace.regex=stats\.page.*

上述示例中的 \ 字符转义了正则表达式中紧随其后的 . 字符。有关如何构建正则表达式的更多信息，请参阅 Java API 文档中有关模式的内容。

默认：""
接受的值：有效的正则表达式

startup.mode.copy.existing.pipeline

Type: string

Description:
An inline array of pipeline operations the connector runs when copying existing data. You can use this setting to filter the source collection and improve the use of indexes in the copying process.

For example, the following setting uses the $match aggregation operator to instruct the connector to copy only documents that contain a closed field with a value of false.

startup.mode.copy.existing.pipeline=[ { "$match": { "closed": "false" } } ]

默认：""
接受值：有效的聚合管道阶段

startup.mode.copy.existing.max.threads

类型：int

说明：
连接器可用于复制数据的最大线程数。

默认值：环境中可用的处理器数量
接受的值：整数

startup.mode.copy.existing.queue.size

类型：int

说明：
连接器在复制数据时可以使用的队列大小。

默认：16000
接受值：整数

startup.mode.copy.existing.allow.disk.use

类型： 布尔值

说明：
当设置为 true 时，连接器会为复制现有聚合使用临时磁盘存储。

默认: true
可接受值: true 或 false

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 主题名称