すべてのソースConnector構成プロパティ

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.supfix

タイプ: 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.区切り文字

型: 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クラスの実装の有効な完全クラス名。

パイプライン

Type: 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"

public. 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

公開. full.document.only.tombzone.on.delete

タイプ: ブール値

説明:
ドキュメントが削除されたときにトンブストーン イベントを返すかどうか。トゥームストーンイベントには、null 値を持つ削除されたドキュメントのキーが含まれています。publish.full.document.onlytrueこの設定は、｛1｝が｛2｝の場合にのみ適用されます。

デフォルト: false
許容値: true または false

change.stream.document.key.as.key

型: ブール値

説明:
ドキュメント キーが存在する場合、ドキュメント キーをソース レコード キーとして使用するか否か。

true に設定すると、コネクターは削除されたドキュメントのキーをトンブストーン イベントに追加します。falseに設定すると、コネクターは再開トークンをトンブストーン イベントのソース キーとして使用します。

デフォルト: true
許容値: true または false

collation

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
受け入れられる値: 整数

pay.await.time.ms

タイプ: long

説明:
空のバッチするを返す前に、新しいデータ変更が変更ストリーム カーソルに報告されるまでサーバーが待機する最大時間（ミリ秒）。

デフォルト: 5000
受け入れられる値: 整数

pay.max.batch.size

型: int

説明:
変更ストリームカーソルをポーリングして新しいデータを取得する際に、1 回のバッチするで読み取るドキュメントの最大数。この設定を使用することで、コネクタ内部でバッファリングされるデータ量を制限できます。

デフォルト: 1000
受け入れられる値: 整数

output.format.key

型: string

説明:
ソースコネクタがキードキュメントを出力するデータ形式を指定します。

デフォルト: json
許容値: bson, json, schema

output.format.value

型: string

説明:
ソースコネクタが値ドキュメントを出力するデータ形式を指定します。

コネクタは、出力データ形式としてProtobufをサポートしています。値を指定し、 Kafka Connect Protobuf Converter を インストールして構成 することで、この形式を有効にできます。schema

デフォルト: 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

Type: string

説明:
SourceRecordのキー ドキュメントのAvroスキーマ定義を指定します。

Avroスキーマの詳細については、 Data 形式ガイド の Avro を参照してください。

デフォルト:

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

許容値: 有効な Avro スキーマ

output.schema.value

Type: 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 の場合、コネクターは存在するすべてのソース データを Change Stream イベントにコピーします。この設定は、非推奨の設定 copy.existing=true と同等です。

ソースコネクタが既存のデータを変換するときに、いずれかのシステムがデータベース内のデータを変更した場合、 MongoDB は最新の変更を反映するために重複した変更ストリームイベントを生成することがあります。データコピーが依存する変更ストリームイベントが冪等であるため、コピーされたデータは結果整合性があり、"少なくとも 1 回" の配信保証に準拠します。

デフォルト: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:

• 10進数形式での UNIXエポックからの整数（例: 30 ）

• 1 秒の精度を持つ 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 Connect フレームワークプロパティをオーバーライドします。

デフォルト: "none"
許容値: "none" または "all"

mongo.errors.log.enable

Type: ブール値

説明:
コネクタがログファイルにエラーを報告するかどうか。

コネクタが検出したすべてのエラーをログに記録するには、これをtrueに設定します。

コネクタが許容しないエラーをログに記録するには、これをfalseに設定します。errors.toleranceまたはmongo.errors.tolerance設定を使用して、コネクタが許容するエラーを指定できます。

重要: このプロパティは、errors.log.enable Connect Frameworkプロパティをオーバーライドします。

デフォルト: false
許容値: trueまたはfalse

mongo.errors.dealterqueue.token.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 トピック名

distinct. partition.name

型: string

説明:
使用するカスタムオフセットパーティション名。このオプションを使用すると、既存のオフセットに無効な再開トークンが含まれている場合に、新しい変更ストリームを開始するようにコネクタに指示できます。

この設定を空白のままにすると、コネクタは接続の詳細に基づいてデフォルトのパーティション名を使用します。

オフセットパーティションに名前を付ける戦略を表示するには、「保存されたオフセットのリセット」を参照してください。

デフォルト: ""
許容値: string。パーティションの名前付けの詳細については、Apache Kafka API ドキュメントの SourceRecord を参照してください。

heartbeat.Interval.ms

型: long

説明:
コネクタがハートビートメッセージを送信するまでの待機時間（ミリ秒）。指定された間隔でソースレコードが公開されない場合、コネクタはハートビートメッセージを送信します。このメカニズムにより、名前空間のデータ量が少ない場合でもコネクタの再開可能性が向上します。

ハートビートメッセージには、postBatchResumeToken データフィールドが含まれています。このフィールドの値には、コネクタが変更ストリームから最後に読み取った MongoDB サーバー oplog エントリが含まれます。

ハートビートメッセージを無効にするには、これを 0 に設定します。

学ぶには、 無効なレジュームトークン ページの 防止 を参照してください。

デフォルト: 0
許容値: 整数

heartbeat.token.name

タイプ: string

説明:
コネクターがハートビート メッセージを公開するトピック名。この機能を有効にするには、heartbeat.interval.ms 設定で正の値を入力する必要があります。

デフォルト: __mongodb_heartbeats
許容値: 有効な Kafka トピック名