4.5.5. Debezium MongoDB コネクターの設定プロパティーを説明します。

name

コネクターの一意名。同じ名前で再登録を試みると失敗します。(このプロパティーはすべての Kafka Connect コネクターに必要です)

connector.class

コネクターの Java クラスの名前。MongoDB コネクターには、常に io.debezium.connector.mongodb.MongoDbConnector の値を使用します。

mongodb.hosts

レプリカセットでの MongoDB サーバーのホスト名とポートのペア ('host' または 'host:port' 形式) のコンマ区切りリスト。リストには、ホスト名とポートのペアを 1 つ含めることができます。mongodb.members.auto.discover を false に設定すると、ホストとポートには、レプリカセット名 (rs0/localhost:27017) を接頭辞として付ける必要があります。

+

mongodb.name

このコネクターが監視するコネクターや MongoDB レプリカセット、またはシャードクラスターを識別する一意の名前。このサーバー名は、MongoDB レプリカセットまたはクラスターから生成される永続化されたすべての Kafka トピックの接頭辞になるため、各サーバーは最大 1 つの Debezium コネクターによって監視される必要があります。名前を設定する文字は、英数字、ハイフン、ドット、アンダースコアのみです。論理名は、このコネクターからレコードを受信する Kafka トピックに名前を付ける際の接頭辞として使用されるため、他のすべてのコネクターで一意である必要があります。

+

mongodb.user

MongoDB への接続時に使用されるデータベースユーザーの名前。これは MongoDB が認証を使用するように設定されている場合にのみ必要です。

mongodb.password

MongoDB への接続時に使用されるパスワード。これは MongoDB が認証を使用するように設定されている場合にのみ必要です。

mongodb.authsource

admin

MongoDB クレデンシャルが含まれるデータベース (認証ソース)。これは、MongoDB が admin 以外の認証データベースで認証を使用するよう設定されている場合に必要です。

mongodb.ssl.enabled

false

コネクターは SSL を使用して MongoDB インスタンスに接続します。

mongodb.ssl.invalid.hostname.allowed

false

SSL が有効な場合、接続フェーズ中に厳密なホスト名のチェックを無効にするかどうかを制御する設定です。true に設定すると、接続で中間者攻撃は阻止されません。

database.include.list

空の文字列

監視するデータベース名と一致する正規表現のコンマ区切りリスト (任意)。database.include.list に含まれていないデータベース名は、監視から除外されます。デフォルトでは、すべてのデータベースが監視されます。database.exclude.list と併用しないでください。

database.exclude.list

空の文字列

監視から除外されるデータベース名と一致する正規表現のコンマ区切りリスト (任意)。database.exclude.list に含まれていないデータベース名が監視の対象となります。database.include.list と併用しないでください。

collection.include.list

空の文字列

監視する MongoDB コレクションの完全修飾 namespace と一致する正規表現のコンマ区切りリスト (任意)。collection.include.list に含まれていないコレクションはすべて監視から除外されます。各識別子の形式は databaseName.collectionName です。デフォルトでは、local および admin データベースにあるコレクションを除くすべてのコレクションがコネクターによって監視されます。collection.exclude.list と併用しないでください。

collection.exclude.list

空の文字列

監視から除外される MongoDB コレクションの完全修飾 namespace と一致する正規表現のコンマ区切りリスト (任意)。collection.exclude.list に含まれていないコレクションはすべて監視されます。各識別子の形式は databaseName.collectionName です。collection.include.list と併用しないでください。

snapshot.mode

Initial

コネクターの起動時にスナップショットを実行する基準を指定します。デフォルトは initial で、オフセットが見つからない場合や oplog/change streams に以前のオフセットが含まれなくなった場合にコネクターがスナップショットを読み取るように指定します。never オプションは、コネクターはスナップショットを使用せずに、ログをの追跡を続行すべきであることを指定します。

capture.mode

change_streams_update_full

MongoDB サーバーからの変更の取り込みに使う方法を指定します。デフォルトは change_streams_update_full で、このコネクターは MongoDB の Change Streams の仕組みで変更を取得し、更新 イベントには完全なドキュメントを含めるように指定されています。change_streams モードでは、同じキャプチャ方法を使用しますが、更新 イベントには完全なドキュメントが含まれません。
oplog モードは、MongoDB の oplog に直接アクセスすることを指定します。これはレガシーな方法なので、新しいコネクターインスタンスには使わないでください。

snapshot.include.collection.list

collection.include.listに指定されたすべてのコレクション

スナップショットを作成する collection.include.list に指定されたスキーマの名前と一致する正規表現のコンマ区切りリスト (任意)。

field.exclude.list

空の文字列

変更イベントメッセージ値から除外される必要があるフィールドの完全修飾名のコンマ区切りリスト (任意)。フィールドの完全修飾名の形式はdatabaseName.collectionName.fieldName.nestedFieldName で、databaseName および collectionName にはすべての文字と一致するワイルドカード (*) が含まれることがあります。

field.renames

空の文字列

イベントメッセージ値のフィールドの名前を変更するために使用されるフィールドの完全修飾置換のコンマ区切りリスト (任意)。フィールドの完全修飾置換の形式は databaseName.collectionName.fieldName.nestedFieldName:newNestedFieldName で、databaseName および collectionName にはすべての文字と一致するワイルドカード (*) が含まれることがあります。コロン (:) は、フィールドの名前変更マッピングを決定するために使用されます。次のフィールドの置換は、リストの前のフィールド置換の結果に適用されるため、同じパスにある複数のフィールドの名前を変更する場合は、この点に注意してください。

tasks.max

1

このコネクターのために作成する必要のあるタスクの最大数。MongoDB コネクターは各レプリカセットに個別のタスクの使用しようとします。そのため、コネクターを単一の MongoDB レプリカセットと使用する場合は、デフォルトを使用できます。MongoDB のシャードクラスターでコネクターを使用する場合、クラスターのシャード数以上の値を指定して、各レプリカセットの作業が Kafka Connect によって分散されるようにすることが推奨されます。

snapshot.max.threads

1

レプリカセットでコレクションの最初の同期を実行するために使用されるスレッドの最大数を指定する正の整数値。デフォルトは 1 です。

tombstones.on.delete

true

削除 イベントの後に廃棄 (tombstone) イベントが続くかどうかを制御します。

true: 削除操作は、削除 イベントと後続の破棄 (tombstone) イベントで表されます。

false: 削除イベントのみ出力されます。

log compaction がトピックで有効になっている場合には、ソースレコードの削除後に廃棄 (tombstone) イベントを出力すると (デフォルト動作)、Kafka は削除された行のキーに関連するすべてのイベントを完全に削除できます。

snapshot.delay.ms

コネクターの起動後、スナップショットを取得するまで待機する間隔 (ミリ秒単位)。
クラスター内で複数のコネクターを開始する際に スナップショットが中断されないようにするために使用でき、コネクターのリバランスが実行される可能性があります。

snapshot.fetch.size

0

スナップショットの実行中に各コレクションから 1 度に読み取る必要があるドキュメントの最大数を指定します。コネクターは、このサイズの複数のバッチでコレクションの内容を読み取ります。
デフォルトは 0 で、サーバーが適切なフェッチサイズを選択することを示します。

schema.name.adjustment.mode

avro

コネクターで使用されるメッセージコンバータとの互換性のために、スキーマ名をどのように調整するかを指定します。設定可能:

max.batch.size

2048

このコネクターの反復処理中に処理される必要があるイベントの各バッチの最大サイズを指定する正の整数値。デフォルトは 2048 です。

max.queue.size

8192

ブロッキングキューが保持できるレコードの最大数を指定する正の整数値。Debezium はデータベースからストリームされたイベントを読み込む際、Kafka に書き込む前にブロッキングキューにイベントを配置します。ブロッキングキューは、コネクターが Kafka に書き込むよりも速くメッセージを取り込む場合、または Kafka が利用できなくなった場合に、データベースから変更イベントを読み込むためのバックプレッシャーを提供することができます。コネクターがオフセットを定期的に記録すると、キューに保持されるイベントは無視されます。max.queue.size の値を、max.batch.size の値よりも大きくなるように設定します。

max.queue.size.in.bytes

0

ブロッキングキューの最大容量をバイト単位で指定する長整数値。デフォルトでは、ブロックキューにはボリューム制限は指定されません。キューが使用できるバイト数を指定するには、このプロパティーを正の long 値に設定します。
max.queue.size も設定されている場合、キューのサイズがどちらかのプロパティーで指定された上限に達すると、キューへの書き込みがブロックされます。例えば、max.queue.size=1000、max.queue.size.in.bytes=5000 と設定した場合、キューに 1000 レコードが入った後、あるいはキュー内のレコードの量が 5000 バイトに達した後、キューへの書き込みがブロックされます。

poll.interval.ms

1000

各反復処理の実行中に新しい変更イベントが表示されるまでコネクターが待機する時間 (ミリ秒単位) を指定する正の整数値。デフォルトは 1000 ミリ秒 (1 秒) です。

connect.backoff.initial.delay.ms

1000

最初に失敗した接続試行の後またはプライマリーが利用できない場合に、プライマリーへの再接続を試行するときの最初の遅延を指定する正の整数値。デフォルトは 1 秒 (1000 ミリ秒) です。

connect.backoff.max.delay.ms

1000

接続試行に繰り返し失敗した後またはプライマリーが利用できない場合に、プライマリーへの再接続を試行するときの最大遅延を指定する正の整数値。デフォルトは 120 秒 (120,000 ミリ秒) です。

connect.max.attempts

16

レプリカセットのプライマリーへの接続を試行する場合の最大失敗回数を指定する正の整数値。この値を越えると、例外が発生し、タスクが中止されます。デフォルトは 16。connect.backoff.initial.delay.ms と connect.backoff.max.delay.ms のデフォルト値では、 20 分強試行した後にのみ失敗します。

mongodb.members.auto.discover

true

'mongodb.hosts' 内のアドレスがクラスターまたはレプリカセットの全メンバーを検出するために使用されるシードであるかどうか (true)、または mongodb.hosts のアドレスをそのまま使用する必要があるかどうか (false) を指定するブール値。デフォルトは true で、MongoDB が プロキシーと面する 場合を除き、すべてのケースで使用する必要があります。

heartbeat.interval.ms

0

ハートビートメッセージが送信される頻度を制御します。
このプロパティーには、コネクターがメッセージをハートビートトピックに送信する頻度を定義する間隔 (ミリ秒単位) が含まれます。これは、コネクターがデータベースから変更イベントを受信しているかどうかを監視するために使用できます。また、長期に渡り変更されるのはキャプチャーされていないコレクションのレコードのみである場合は、ハートビートメッセージを利用する必要があります。このような場合、コネクターはデータベースからの oplog/change streams の読み取りを続行しますが、変更メッセージを Kafka に出力しないため、オフセットの更新が Kafka にコミットされません。これにより、oplog ファイルがローテーションされますが、コネクターはこれを認識しないため、再起動時に一部のイベントが利用できなくなり、最初のスナップショットの再実行が必要になります。

このプロパティーを 0 に設定して、ハートビートメッセージが全く送信されないようにします。
デフォルトでは無効にされています。

heartbeat.topics.prefix

__debezium-heartbeat

ハートビートメッセージが送信されるトピックの命名を制御します。
トピックは、<heartbeat.topics.prefix>.<server.name> パターンに従って名前が付けられます。

sanitize.field.names

コネクター設定が、Avro を使用するように key.converter または value.converter パラメーターを明示的に指定する場合は true です。それ以外の場合のデフォルトは false です。

Avro の命名要件に準拠するためにフィールド名がサニタイズされるかどうか。

skipped.operations

ストリーミング中にスキップされる操作タイプのコンマ区切りリスト。操作には、c (挿入/作成)、u (更新)、および d (削除) が含まれます。デフォルトでは、操作はスキップされません。

snapshot.collection.filter.overrides

スナップショットに含まれるコレクション項目を制御します。このプロパティーはスナップショットにのみ影響します。databaseName.collectionName の形式でコレクション名のコンマ区切りリストを指定します。

指定する各コレクションに対して、別の設定プロパティー (snapshot.collection.filter.overrides. databaseName.collectionName) も指定します。たとえば、他の設定プロパティーの名前は snapshot.collection.filter.overrides.customers.orders などです。このプロパティーは、スナップショットで必要なアイテムのみを取得する有効なフィルター式に設定します。コネクターがスナップショットを実行すると、フィルター式と一致する項目のみを取得します。

provide.transaction.metadata

false

true に設定すると、Debezium はトランザクション境界でイベントを生成し、トランザクションメタデータでデータイベントエンベロープをエンリッチします。

詳細は、トランザクションメタデータ を参照してください。

transaction.topic

${database.server.name}.transaction

コネクターがトランザクションのメタデータメッセージを送信するトピックの名前を制御します。プレースホルダ ${database.server.name} は、コネクターの論理名として使用できます (「Debezium MongoDB コネクターでレプリカセットおよびシャードクラスターに論理名を使用する方法」参照)。デフォルトは ${database.server.name}.transaction で、たとえば dbserver1.transaction のような名前になります。

retriable.restart.connector.wait.ms

10000 (10 秒)

再試行可能なエラーが発生した後にコネクターを再起動するまで待機する時間 (ミリ秒単位)。

mongodb.poll.interval.ms

30000

コネクターが新規、削除、または変更したレプリカセットをポーリングする間隔。

mongodb.connect.timeout.ms

10000 (10 秒)

新しい接続試行が中断されるまでドライバーが待機する時間 (ミリ秒単位)。

mongodb.socket.timeout.ms

0

ソケットでの送受信がタイムアウトするまでにかかる時間 (ミリ秒単位)。0 の値は、この動作を無効にします。

mongodb.server.selection.timeout.ms

30000 (30 秒)

ドライバーがタイムアウトし、エラーが出力される前に、サーバーが選択されるまでドライバーが待つ時間 (ミリ秒単位)。

cursor.max.await.time.ms

0

実行タイムアウトの例外を発生させる前に、oplog/Change Streams カーソルが結果を生成するのを待つ最大期間 (ミリ秒単位) を指定します。値 0 は、サーバー/ドライバーのデフォルト待機タイムアウトを使用することを示します。

signal.data.collection

デフォルトなし

シグナルをコネクターへの送信に使用されるデータコレクションの完全修飾名 。コレクション名を指定するには、次の形式を使用します。
<databaseName>.<collectionName>
シグナリングは、Debezium MongoDB コネクターの技術プレビュー機能です。

incremental.snapshot.chunk.size

1024

増分スナップショットチャンク中にコネクターがフェッチしてメモリーに読み込むドキュメントの最大数です。スナップショットは、サイズが大きいスナップショットの場合にはクエリーが少なくなるため、チャンクサイズを増やすと効率が上がります。ただし、チャンクサイズが大きい場合には、スナップショットデータのバッファーにより多くのメモリーが必要になります。チャンクサイズは、環境で最適なパフォーマンスを発揮できる値に、調整します。
増分ナップショットは、Debezium MongoDB コネクターのテクノロジープレビュー機能です。