Red Hat Summit Red Hat Summitサポートコンソール開発者トライアルの開始お問い合わせ

7.10. 送信トレイパターンを使用する Debezium MongoDB コネクターの設定

注記

この SMT は、Debezium MongoDB コネクターでのみ使用されます。リレーショナルデータベースに送信ボックスイベントルーター SMT を使用する方法は、送信トレイ (Outbox) イベントルーター を参照してください。

送信トレイパターンを使用することで、複数の (マイクロ) サービス間で安全かつ確実にデータを交換することができます。送信トレイパターンの実装により、サービスの内部状態 (通常はそのデータベースに永続化される) と同じデータを必要とするサービスで使用されるイベントの状態との間に不整合が生じるのを防ぐことができます。

Debezium アプリケーションに送信トレイパターンを実装するには、Debezium コネクターを以下のように設定します。

  • 送信トレイコレクションの変更をキャプチャーする
  • Debezium MongoDB 送信トレイイベントルーター単一メッセージ変換 (SMT) を適用する

MongoDB 送信トレイ SMT を適用するように設定された Debezium コネクターは、送信トレイコレクションで生じた変更だけをキャプチャーする必要があります。詳細は、変換を選択的に適用するオプション を参照してください。

コネクターが複数の送信トレイコレクションの変更をキャプチャーすることができるのは、それぞれの送信トレイコレクションが同じ構造を持つ場合に限ります。

注記

この SMT を使用するには、実際のビジネスコレクションの操作と送信トレイコレクションへの挿入を、ビジネスコレクションおよび送信トレイコレクション間にデータの不整合が発生するのを回避するために、MongoDB 4.0 以降でサポートされているマルチドキュメントトランザクションの一部として実行する必要があります。将来の更新では、既存のデータを更新し、マルチドキュメントトランザクションなしで ACID トランザクションに送信トレイイベントを挿入できるようにするために、送信トレイイベントを、独立した送信トレイコレクションとしてではなく、既存コレクションのサブドキュメントとして保存するための追加の設定をサポートする予定です。

送信トレイパターンの詳細は、Reliable Microservices Data Exchange With the Outbox Pattern を参照してください。

詳細は以下のセクションを参照してください。

7.10.1. Debezium MongoDB 送信トレイメッセージの例
リンクのコピー

Debezium MongoDB 送信トレイイベントルーター SMT の設定方法を理解するには、以下の Debezium 送信トレイメッセージの例を検討してください。 

# Kafka Topic: outbox.event.order
# Kafka Message key: "b2730779e1f596e275826f08"
# Kafka Message Headers: "id=596e275826f08b2730779e1f"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}"
}
Copy to Clipboard Toggle word wrap

MongoDB 送信トレイイベントルーター SMT を適用するように設定された Debezium コネクターは、次の例で示すとおり、raw Debezium 変更イベントメッセージを変換して上記のメッセージを生成します。 

# Kafka Message key: { "id": "{\"$oid\": \"596e275826f08b2730779e1f\"}" }
# Kafka Message Headers: ""
# Kafka Message Timestamp: 1556890294484
{
  "patch": null,
  "after": "{\"_id\": {\"$oid\": \"596e275826f08b2730779e1f\"}, \"aggregateid\": {\"$oid\": \"b2730779e1f596e275826f08\"}, \"aggregatetype\": \"Order\", \"type\": \"OrderCreated\", \"payload\": {\"_id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}}",
  "source": {
    "version": "2.7.3.Final",
    "connector": "mongodb",
    "name": "fulfillment",
    "ts_ms": 1558965508000,
    "ts_us": 1558965508000000,
    "ts_ns": 1558965508000000000,
    "snapshot": false,
    "db": "inventory",
    "rs": "rs0",
    "collection": "customers",
    "ord": 31,
    "h": 1546547425148721999
  },
  "op": "c",
  "ts_ms": 1556890294484,
  "ts_us": 1556890294484452,
  "ts_ns": 1556890294484452697,
}
Copy to Clipboard Toggle word wrap

この Debezium 送信トレイメッセージの例は、デフォルトの送信トレイイベントルーター設定 に基づいています。ここでは、送信トレイコレクション構造および集約に基づくイベントルーティングを想定しています。動作をカスタマイズするために、送信トレイイベントルーター SMT にはさまざまな 設定オプション が用意されています。

デフォルトの MongoDB 送信トレイイベントルーター SMT 設定を適用するには、送信トレイコレクションに以下のフィールドがなければなりません。 

{
  "_id": "objectId",
  "aggregatetype": "string",
  "aggregateid": "objectId",
  "type": "string",
  "payload": "object"
}
Copy to Clipboard Toggle word wrap
Expand
表7.10 要求される送信トレイコレクションフィールドの説明
フィールド結果

id

イベントの一意の ID が含まれます。送信トレイメッセージでは、この値はヘッダーです。たとえば、重複するメッセージを削除するために、この ID を使用することができます。

イベントの一意の ID を別の送信トレイコレクションフィールドから取得するには、コネクター設定で collection.field.event.id SMT オプションを設定します。

aggregatetype

コネクターが送信トレイメッセージを出力するトピックの名前に SMT が追加する値が含まれます。デフォルトの動作では、この値は route.topic.replacement SMT オプションのデフォルトの ${routedByValue} 変数を置き換えます。

たとえば、デフォルトの設定では、route.by.field SMT オプションは aggregatetype に、route.topic.replacement SMT オプションは outbox.event.${routedByValue} に設定されます。アプリケーションが 2 つのドキュメントを送信トレイコレクションに追加するとします。最初のドキュメントでは、aggregatetype フィールドの値は customers です。2 番目のドキュメントでは、aggregatetype フィールドの値は orders です。コネクターは、最初のドキュメントを outbox.event.customers トピックに送信します。コネクターは、2 番目のドキュメントを outbox.event.orders トピックに送信します。

別の送信トレイコレクションフィールドからこの値を取得するには、コネクター設定で route.by.field SMT オプションを設定します。

aggregateid

ペイロードの ID を提供するイベントのキーが含まれます。SMT は、この値を出力される送信トレイメッセージのキーとして使用します。これは、Kafka パーティションで正しい順序を維持するのに重要です。

別の送信トレイコレクションフィールドからイベントキーを取得するには、コネクター設定で collection.field.event.key SMT オプションを設定します。

payload

送信トレイ変更イベントの表現。デフォルトの構造は JSON です。デフォルトでは、Kafka のメッセージ値は payload 値のみで構成されます。ただし、送信トレイイベントで追加のフィールドが含まれるように設定されている場合、Kafka メッセージ値にはペイロードと追加フィールドの両方のエンベロープエンプトが含まれ、各フィールドは個別に表されます。詳細は、追加フィールドでのメッセージ出力 を参照してください。

別の送信トレイコレクションフィールドからイベントペイロードを取得するには、コネクター設定で collection.field.event.payload SMT オプションを設定します。

追加のカスタムフィールド

送信トレイコレクションの追加フィールドは、ペイロードセクション内に、またはメッセージヘッダーとして、送信トレイイベントに追加 できます。

一例として、イベントを分類または整理するのに役立つ、ユーザー定義の値を伝えるフィールド eventType などがあります。

7.10.3. 基本的な Debezium MongoDB 送信トレイイベントルーター SMT 設定
リンクのコピー

送信トレイパターンをサポートするように Debezium MongoDB コネクターを設定するには、outbox.MongoEventRouter SMT を設定します。SMT のデフォルトの動作を取得するには、次の例のように、オプションを指定せずにコネクター設定に追加します。 

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
Copy to Clipboard Toggle word wrap

設定のカスタマイズ

コネクターは、多くの種類のイベントメッセージ (ハートビートメッセージ、廃棄メッセージ、またはトランザクションの変更に関するメタデータメッセージ) を発行する場合があります。送信トレイコレクションで発生したイベントだけに変換を適用するには、これらのイベントのみを対象に 変換を選択して適用する SMT 述語ステートメント を定義します。

7.10.4. MongoDB 送信トレイイベントルーター変換を選択的に適用するオプション
リンクのコピー

データベースの変更が発生したときに Debezium コネクターが出力する変更イベントメッセージの他に、コネクターはハートビートメッセージなど、他のタイプのメッセージとスキーマ変更およびトランザクションに関するメタデータメッセージも出力します。これらの他のメッセージの構造は、SMT が処理するように設計された変更イベントメッセージの構造とは異なるため、目的のデータ変更メッセージのみを処理するようにコネクターを SMT を選んで適用することが推奨されます。以下の方法のいずれかを使用して、SMT を選んで適用するようにコネクターを設定できます。

7.10.5. Debezium MongoDB 送信トレイメッセージでペイロードフォーマットとして Avro を使用
リンクのコピー

MongoDB 送信ボックスイベントルーター SMT は、任意のペイロード形式をサポートします。送信トレイコレクションの payload フィールド値は透過的に渡されます。JSON を使用する代わりに、Avro を使用することもできます。これは、メッセージフォーマットの管理や、送信トレイイベントスキーマの後方互換性を維持した進化の確保に役立ちます。

送信トレイメッセージペイロード用にソースアプリケーションがどのように Avro フォーマットのコンテンツを生成するかは、このドキュメントの範囲外です。1 つの可能性として、KafkaAvroSerializer クラスを利用して GenericRecord インスタンスをシリアライズできます。Kafka メッセージの値が正確な Avro バイナリーデータとなるようにするには、以下の設定をコネクターに適用します。 

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
value.converter=io.debezium.converters.ByteArrayConverter
Copy to Clipboard Toggle word wrap

デフォルトでは、payload フィールドの値 (Avro データ) が唯一のメッセージ値となります。値変換器として ByteArrayConverter を設定すると、payload フィールドの値がそのまま Kafka メッセージの値に伝搬されます。

これは、他の SMT に推奨される BinaryDataConverter とは異なることに注意してください。これは、MongoDB が内部でバイト配列を保存する際に取っているアプローチが異なるためです。

ハートビート、トランザクションメタデータ、またはスキーマ変更イベントを出力するように Debezium コネクターを設定することができます (サポートはコネクターによって異なります)。これらのイベントは ByteArrayConverter でシリアライズできないため、コンバーターがこれらのイベントのシリアライズ方法を認識できるように追加の設定を指定する必要があります。例として、以下の設定では、スキーマがない状態で Apache Kafka JsonConverter を使用することを示しています。 

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
value.converter=io.debezium.converters.ByteArrayConverter
value.converter.delegate.converter.type=org.apache.kafka.connect.json.JsonConverter
value.converter.delegate.converter.type.schemas.enable=false
Copy to Clipboard Toggle word wrap

委譲 Converter 実装は delegate.converter.type オプションで指定します。コンバーターで追加の設定オプションが必要な場合は (例: 上記の schemas.enable=false を使用したスキーマの無効化)、それらを指定することもできます。

7.10.6. Debezium MongoDB 送信トレイメッセージへの追加フィールドの出力
リンクのコピー

送信トレイコレクションに含まれるフィールドの値を、出力される送信トレイメッセージに追加することができます。たとえば、aggregatetype フィールドに purchase-order という値を持ち、eventType というフィールドに order-created および order-shipped という値を持つ送信トレイコレクションを考えてみましょう。field:placement:alias の構文でフィールドを追加できる。

placement に許可されている値は、header、envelope、partition です。

eventType フィールドの値を送信トレイメッセージのヘッダーに出力するには、以下のような SMT を設定します。 

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.collection.fields.additional.placement=eventType:header:type
Copy to Clipboard Toggle word wrap

結果は、type がキーとして Kafka メッセージのヘッダー、および eventType 列の値はその値になります。

eventType フィールドの値を送信トレイメッセージのエンベロープに出力するには、以下のような SMT を設定します。 

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.collection.fields.additional.placement=eventType:envelope:type
Copy to Clipboard Toggle word wrap

送信メッセージをどのパーティションで生成するかを制御するには、SMT を次のように設定します。 

transforms=outbox,...
transforms.outbox.type=io.debezium.transforms.outbox.EventRouter
transforms.outbox.collection.fields.additional.placement=partitionField:partition
Copy to Clipboard Toggle word wrap

なお、partition 配置については、エイリアスを追加しても効果はありません。

7.10.7. JSON としてエスケープされた JSON 文字列の拡張
リンクのコピー

デフォルトでは、Debezium 送信トレイメッセージの payload は文字列として表されます。文字列の元のソースが JSON 形式の場合、次の例に示すように、結果の Kafka メッセージはエスケープシーケンスを使用して文字列を表します。 

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=596e275826f08b2730779e1f"
# Kafka Message Timestamp: 1556890294484
{
  "{\"id\": {\"$oid\": \"da8d6de63b7745ff8f4457db\"}, \"lineItems\": [{\"id\": 1, \"item\": \"Debezium in Action\", \"status\": \"ENTERED\", \"quantity\": 2, \"totalPrice\": 39.98}, {\"id\": 2, \"item\": \"Debezium for Dummies\", \"status\": \"ENTERED\", \"quantity\": 1, \"totalPrice\": 29.99}], \"orderDate\": \"2019-01-31T12:13:01\", \"customerId\": 123}"
}
Copy to Clipboard Toggle word wrap

メッセージの内容をデプロイメントし、エスケープされた JSON を元のエスケープされていない JSON 形式に変換するように、送信トレイイベントルーターを設定できます。変換された文字列では、コンパニオンスキーマは元の JSON ドキュメントから推定されます。次の例は、結果の Kafka メッセージでデプロイメントされた JSON を示しています。 

# Kafka Topic: outbox.event.order
# Kafka Message key: "1"
# Kafka Message Headers: "id=596e275826f08b2730779e1f"
# Kafka Message Timestamp: 1556890294484
{
  "id": "da8d6de63b7745ff8f4457db", "lineItems": [{"id": 1, "item": "Debezium in Action", "status": "ENTERED", "quantity": 2, "totalPrice": 39.98}, {"id": 2, "item": "Debezium for Dummies", "status": "ENTERED", "quantity": 1, "totalPrice": 29.99}], "orderDate": "2019-01-31T12:13:01", "customerId": 123
}
Copy to Clipboard Toggle word wrap

変換時の文字列変換を有効にするには、collection.expand.json.payload の値を true に設定し、次の例に示すように StringConverter を使用します。 

transforms=outbox,...
transforms.outbox.type=io.debezium.connector.mongodb.transforms.outbox.MongoEventRouter
transforms.outbox.collection.expand.json.payload=true
value.converter=org.apache.kafka.connect.storage.StringConverter
Copy to Clipboard Toggle word wrap

7.10.8. 送信トレイイベントルーター変換設定用のオプション
リンクのコピー

次の表で、送信トレイイベントルーター SMT に指定することのできるオプションを説明します。表の グループ 列は、Kafka の設定オプションクラスを示しています。

Expand
表7.11 送信トレイイベントルーター SMT 設定オプションの説明
オプションデフォルトグループ説明

collection.op.invalid.behavior

warn

コレクション

送信トレイコレクションも更新操作がある場合の SMT の動作を決定します。設定可能な値は以下のとおりです。

  • warn: SMT はログに警告を記録し、次の送信トレイコレクションドキュメントに進みます。
  • error: SMT はログにエラーを記録し、次の送信トレイコレクションドキュメントに進みます。
  • fatal: SMT はログにエラーを記録し、コネクターは処理を停止します。

送信トレイコレクションのすべての変更は、挿入または削除操作であると想定されます。つまり、送信トレイコレクションはキューとして機能します。送信トレイコレクション内のドキュメントの更新は許可されていません。SMT は、送信トレイコレクションの削除操作を自動的に除外します (進行中の送信トレイイベントが削除されるため)。

collection.field.event.id

_id

コレクション

一意のイベント ID が含まれる送信トレイコレクションフィールドを指定します。この ID は、出力されるイベントのヘッダーの id キーに保存されます。

collection.field.event.key

aggregateid

コレクション

イベントキーが含まれる送信トレイコレクションフィールドを指定します。このフィールドに値が含まれる場合、SMT はその値を出力される送信トレイメッセージのキーとして使用します。これは、Kafka パーティションで正しい順序を維持するのに重要です。

collection.field.event.timestamp

 

コレクション

デフォルトでは、出力される送信トレイメッセージのタイムスタンプは、Debezium イベントのタイムスタンプです。送信トレイメッセージで別のタイムスタンプを使用するには、このオプションを出力される送信トレイメッセージに使用するタイムスタンプが含まれる送信トレイコレクションフィールドに設定します。

collection.field.event.payload

payload

コレクション

イベントペイロードが含まれる送信トレイコレクションフィールドを指定します。

collection.expand.json.payload

false

コレクション

String ペイロードの JSON 拡張を実行するかどうかを指定します。コンテンツが見つからなかった場合や、解析エラーの場合は、コンテンツは "そのまま" として保持されます。

詳細は、エスケープされた json の拡張 セクションを参照してください。

collection.fields.additional.placement

 

コレクション、エンベロープ

送信トレイメッセージのヘッダーまたはエンベロープに追加する 1 つまたは複数の送信トレイコレクションフィールドを指定します。ペアのコンマ区切りリストを指定します。それぞれのペアで、フィールドの名前および値をヘッダーとエンベロープのどちらに含めるかを指定します。ペア内の値はコロンで区切ります。以下に例を示します。

id:header,my-field:envelope

フィールドのエイリアスを指定するには、3 番目の値としてエイリアスが含まれるトリオを指定します。以下にフィールドを示します。

id:header,my-field:envelope:my-alias

2 番目の値は配置で、常に header または envelope でなければなりません。

設定例は、Debezium 送信トレイメッセージへの追加フィールドの出力 に記載されています。

collection.field.event.schema.version

 

コレクション、スキーマ

このオプションを設定すると、Kafka Connect スキーマ Javadoc で説明されているように、その値がスキーマバージョンとして使用されます。

route.by.field

aggregatetype

ルーター

送信トレイコレクションのフィールドの名前を指定します。デフォルトでは、このフィールドで指定された値が、コネクターが送信トレイメッセージを出力するトピックの名前の一部になります。例を 要求される送信トレイコレクションの説明 に示します。

route.topic.regex

(?<routedByValue>.*)

ルーター

送信トレイ SMT が RegexRouter で送信トレイコレクションドキュメントに適用する正規表現を指定します。この正規表現は、route.topic.replacement SMT オプションの設定の一部です。

このプロパティーがデフォルトの Router 値に設定されている場合、SMT は、route.topic.replacement プロパティーに設定されているデフォルトの ${routedByValue} 変数を、route.by.field プロパティーに指定された値に置き換えます。

route.topic.replacement

outbox.event​.${routedByValue}

ルーター

コネクターが送信トレイメッセージを出力するトピックの名前を指定します。デフォルトのトピック名には、文字列 outbox.event. が接頭辞として付加され、その後に、送信ボックスコレクションドキュメントに割り当てられた Aggregatetype フィールドの値が続きます。

たとえば、aggregatetype の値が customers の場合、コネクターはトピック outbox.event.customers に送信ボックスメッセージを送信します。

宛先トピックのデフォルト名を変更するには、次のいずれかのアクションを実行します。

route.tombstone.on.empty.payload

false

ルーター

空または null のペイロードによってコネクターがトゥームストーンイベントを出力するかどうかを示します。

トップに戻る
Red Hat logoGithubredditYoutubeTwitter

詳細情報

試用、購入および販売

コミュニティー

Red Hat ドキュメントについて

Red Hat をお使いのお客様が、信頼できるコンテンツが含まれている製品やサービスを活用することで、イノベーションを行い、目標を達成できるようにします。 最新の更新を見る.

多様性を受け入れるオープンソースの強化

Red Hat では、コード、ドキュメント、Web プロパティーにおける配慮に欠ける用語の置き換えに取り組んでいます。このような変更は、段階的に実施される予定です。詳細情報: Red Hat ブログ.

会社概要

Red Hat は、企業がコアとなるデータセンターからネットワークエッジに至るまで、各種プラットフォームや環境全体で作業を簡素化できるように、強化されたソリューションを提供しています。

Theme

© 2025 Red Hat