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

connection.uri

必須

タイプ: string

説明:
MongoDBインスタンスまたはクラスターに接続するための URI接続文字列。詳しくは、

「 MongoDBへの接続 」を参照してください。

重要: connection.uri設定で認証情報が公開されないようにするには、ConfigProvider を使用して適切な構成パラメーターを設定します。

デフォルト:mongodb://localhost:27017,localhost:27018,localhost:27019
許容値: MongoDB URI接続文字列

database

タイプ: string

説明:
変更を監視するデータベースの名前。設定されていない場合、コネクタはすべてのデータベースの変更を監視します。デフォルト:

""
許容値: 1 つのデータベース名

コレクション

タイプ: string

説明:
変更を監視するデータベース内のコレクションの名前。設定されていない場合、コネクタはすべてのコレクションの変更を監視します。重要:

database構成が"" に設定されている場合、コネクタは 設定を無視します。デフォルト:collection

""
許容値: 単一のコレクション名

server.api.version

タイプ: string

説明:
MongoDBクラスターで使用する Stable APIバージョン。 Stable APIとそれをサポートするMongoDBサーバーのバージョンの詳細については、 Stable APIガイド を参照してください。デフォルト:

""
使用可能な値: 空の文字列または有効な Stable APIバージョン。

server.api.deprecationErrors

タイプ:ブール値説明:
trueに設定すると、 コネクタ が宣言された Stable APIバージョンで非推奨のMongoDBインスタンスで コマンドを呼び出すと、例外が発生します。

server.api.version構成オプションを使用してAPIバージョンを設定できます。 Stable APIの詳細については、 Stable APIに関するMongoDB のマニュアル エントリを参照してください。デフォルト:

false
許容値:true またはfalse

server.api.strict

タイプ:ブール値説明:
true

に設定すると、server.api.version コネクタ が宣言された Stable APIバージョンでカバーされていないMongoDBインスタンス上のコマンドを呼び出すと、例外が発生します。 構成オプションを使用してAPIバージョンを設定できます。 Stable APIの詳細については、 Stable APIに関するMongoDB のマニュアル エントリを参照してください。デフォルト:

false
許容値:true またはfalse

topic.prefix

タイプ: string


説明:コネクタが変更ストリームイベントを公開する宛先Kafkaトピック名の最初の部分を指定します。宛先トピック名は、topic.prefix 値とそれに続くデータベースとコレクション名で構成され、 プロパティで指定された値で区切られています。詳細については、topic.separator

トピックの命名プレフィックス

""
の例を参照してください。デフォルト: 許容値: "."、"-"、"_" などの ASCII 英数字文字で構成される string

topic.supfix

タイプ: string


説明:コネクタが変更ストリームイベントを公開する宛先Kafkaトピック名の最後の部分を指定します。宛先トピック名は、データベースとコレクション名、その後にtopic.suffix 値を構成し、 プロパティで指定された値で区切った構成になっています。詳細については、topic.separator

トピックの命名サフィックス

""
の例を参照してください。デフォルト: 許容値: "."、"-"、"_" などの ASCII 英数字文字で構成される string

topic.namespace.map

タイプ: string


説明:変更ストリームドキュメントの名前空間とトピック名の間のJSONマッピングを指定します。複雑なマッピングを指定するには、

をtopic.namespace.map プロパティに使用します。このプロパティは、正規表現とワイルドカードの一致をサポートします。これらの動作と表示例について詳しくは、「

トピック名前空間マップ

」を参照してください。デフォルト:""
許容値: 有効なJSONオブジェクト

topic.区切り文字

タイプ: string 説明:トピックの名前を作成するために使用される値を連結するためにコネクタが使用する 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=-

重要: topic.separatorプロパティを使用する場合、 プロパティtopic.namespace.map の定義方法には影響しないことに注意してください。topic.namespace.map プロパティはMongoDB名前空間を使用します。この名前空間は、データベースとコレクション名を区切るために常に 文字で指定する必要があります。デフォルト:.

"."
許容値: string

topic.mapper

タイプ: string




説明:com.mongodb.kafka.connect.source.topic.mapping.DefaultTopicMapper
カスタムトピックマッピング ロジックを定義するJavaクラス。デフォルト: 許容値: ToicMapperクラスの実装の有効な完全なクラス名。

パイプライン

タイプ: string


説明:変更ストリームで実行する集計パイプラインの配列。この設定は、fullDocument フィールドではなく、変更ストリームイベントドキュメントに対して構成する必要があります。

（例: ）。

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

その他の例については、次を参照してください。

• 変更イベントをフィルタリングするためのパイプラインのカスタマイズ
• 複数のソースの変更をリッスンする

デフォルト:"[]"
許容値: 有効な集計パイプラインステージ

change.stream. full.document

タイプ: string






説明:updateLookup アップデート操作時に変更ストリームが返す値を決定します。デフォルト設定では、元のドキュメントと更新されたドキュメントとの差が返されます。

whenAvailable設定は、元のドキュメントと更新されたドキュメントとの差、および更新後の特定の点での更新されたドキュメント全体のコピーを返します。

required設定は更新されたドキュメントを返します（利用可能な場合）。

設定は更新されたドキュメントを返し、ドキュメントが利用できない場合はエラーを発生させます。この変更ストリームオプションの機能の詳細については、 MongoDBマニュアルの「 更新操作のための完全なドキュメントの検索

」を参照してください。デフォルト:""
許容値:"" 、"updateLookup" 、"whenAvailable" 、または"required"

change.stream.show.expanded.events

タイプ:ブール値値

説明:
変更ストリームが
createIndexes イベントや
dropIndexes
イベントなどの DDL イベントを通知するかどうかを決定します。詳しくは、 showExpandedEvents を参照してください。この設定は、アップデート


updateDescription.disambiguatedPathsイベントで
を表示するために必要です。これにより、あいまいなフィールドを含む変更が明確になります。詳しくは、
disabledPaths を参照してください。デフォルト:

false
許容値:true またはfalse

change.stream. full.document.Before.change

タイプ: string

説明:
アップデート操作時に変更ストリームが返す変更前のドキュメントを構成します。変更前のイメージは、既存のデータのコピー中に公開されたソース レコードでは使用できず、変更前のイメージ構成はコピーに影響しません。変更前イメージを有効にするようにコレクションを構成する方法については、

MongoDBマニュアルの「 Change Streams とドキュメントの変更前イメージおよび変更後イメージ

」を参照してください。デフォルト設定では、ドキュメントの変更前のイメージは非表示にされます。

whenAvailable設定では、置換、アップデート、または削除される前に、ドキュメントが利用可能な場合、変更前のイメージが返されます。

required設定では、ドキュメントの変更前のイメージが返され、利用できない場合はエラーが発生します。デフォルト:

""
許容値:"" または"whenAvailable" または"required"

public. full.document.only

タイプ:ブール値値

説明:
fullDocumentfullDocumentfullDocument更新イベントによって生成された変更ストリームイベントドキュメントから フィールドのみを返すかどうか。 フィールドには更新されたドキュメントの最新バージョンが含まれます。 フィールドの詳細については、サーバー マニュアルの更新イベントを参照してください。

trueに設定すると、コネクタは 設定を上書きし、それを に設定して、 フィールドに更新されたドキュメントが含まれるようにします。デフォルト:change.stream.full.document updateLookupfullDocument

false
許容値:true またはfalse

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

タイプ:ブール値値

説明:
ドキュメントが削除されたときにトゥームストーンnull イベントを返すかどうか。 Toms の イベントには、 値を持つ削除されたドキュメントのキーが含まれます。この設定は、publish.full.document.only が の場合にのみ適用されます。デフォルト:true

false
許容値:true またはfalse

change.stream.document.key.as.key

タイプ:ブール値値


説明:ドキュメントキーが存在する場合、ソースレコードキーにドキュメントキーを使用するかどうか。

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

true
許容値:true またはfalse

collation

タイプ: string


説明:変更ストリームによって返されるドキュメントにMongoDB が適用する言語固有の順序付けルールを指定するJSON照合ドキュメント。デフォルト:

""
許容値: 有効な照合JSONドキュメント

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 をサポートします。この形式を有効にするには、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 のキードキュメントの Atlasスキーマ定義を指定します。

Avro スキーマの詳細については、 データ形式ガイドの Avro を参照してください。デフォルト：

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

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

output.schema.value

タイプ: string

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

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 と同じです。

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

デフォルト:latest
許容値:latest 、timestamp 、copy_existing

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

タイプ: string

説明:
startup.mode=timestampの場合にのみ有効になります。変更ストリームの開始点を指定します。

Change Stream パラメーターの詳細については、 MongoDBマニュアルの $changeStream（集計）を参照してください。デフォルト:

""
許容値:

• 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 データベース内の「ページ」で始まるコレクションと一致します。

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

\上記の例では、. 文字は、正規式内のそれに続く 文字をエスケープします。正規表現の構築方法の詳細については、 Java APIドキュメントの パターン

を参照してください。デフォルト:""
許容値: 有効な正規式

startup.mode. copy.existing. pipeline

タイプ: string

説明:
既存のデータをコピーするときにコネクタが実行するパイプライン操作のインライン配列。この設定を使用すると、ソースコレクションをフィルタリングし、コピー

プロセスにおけるインデックスの使用を改善できます。例、次の設定では $match集計子を使用して、 closedの値を持つ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 Frameworkプロパティを上書きします。デフォルト:

"none"
許容値:"none" または"all"

mongo.errors.log.enable

タイプ:ブール値値 説明:コネクタがログファイルのエラーを報告するかどうか。コネクタが発生したすべてのエラーをログには、これを




true

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

設定を使用して、コネクタが許容するエラーを指定できます。重要: このプロパティはエラーを上書きします。 ログ.enable Connect

Frameworkプロパティ.デフォルト:false
許容値:true またはfalse

mongo.errors.dealterqueue.token.name

タイプ: string 説明:デッドレターキュー




(DLQ)として使用するトピックの名前。値を指定すると、コネクタは無効なメッセージを拡張JSON string としてデッドレターキュー

(DLQ)トピックに書込みます。この設定を空白のままにすると、コネクタはトピックに無効なメッセージを書込みません。

重要: errors.tolerancemongo.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トピック名