2.5. 新機能および改善点

MySQL および PostgreSQL の Debezium コネクターを設定して、セキュアな SSL 接続を使用できます。MySQL コネクターの場合、database.ssl.mode プロパティーを設定しえｔセキュアなな接続の使用を指定します。PostgreSQL コネクターの場合は、database.sslmode プロパティーを設定します。

Debezium 2.3.4 以降、これらの設定オプションには新しいデフォルト値が含まれています。MySQL の場合、以前のデフォルト値の disabled に代わって、database.ssl.mode のデフォルト値が preferred になりました。PostgreSQL の場合、database.sslmode のデフォルト値は、以前のデフォルト値 disable に代わって、prefer になりました。新しいデフォルト設定をもとに、コネクターは、データベースへの接続を開始するときに、まず、セキュアで暗号化された接続を確立しようとします。安全な接続が利用できない場合、コネクターは、別途設定されていない限り、セキュアでない接続を使用するようにフォールバックします。

Debezium はトピック名とスキーマ名を生成するときに、スキーマレジストリーの命名規則との互換性を確保するために、名前に含まれる ASCII 以外の文字を置き換えます。以前のリリースでは、Debezium は ASCII 以外の文字をアンダースコア文字 (_) に置き換えていました。ただし、場合によっては ASCII 以外の文字を置き換えた後、Debezium が 2 つのトピックまたはスキーマに対して生成する名前が、大文字と小文字を除いて同一になる可能性があり、他の問題が発生する可能性があります。

最も互換性のある方法でこれに対処するために、Debezium はストラテジーベースのアプローチを使用して文字を独自にマップするようになりました。この新しいアプローチによる影響の 1 つとして、Debezium が sanitize.field.names 設定プロパティーがサポートされなくなっています。sanitize.field.names プロパティーの代わりに、テーブルまたはコレクションに使用する規則と互換性のある命名ストラテジーを指定するための新しいオプションが使用できるようになりました。

Debezium によるスキーマおよびフィールド名の生成方法を指定するには、以下のプロパティーを設定します。

前述の各プロパティーに対して、次のいずれかの値を設定できます。

Debezium が挿入、更新、および削除イベントに対して発行する変更イベントレコードには、source 情報ブロックを含むペイロードが含まれています。Oracle コネクターの場合、ソース情報ブロックには、変更の SQL シーケンス番号を表す特別な ssn フィールドが含まれています。

ソースデータベースの ssn フィールドの値が INT32 データ型の最大値 (2,147,483,647) を超える場合があります。Debezium は、より大きな値を許可するため、ssn フィールドにデータ型 INT64 を割り当てるようになり、フィールドの最大値が 9,223,372,036,854,775,807 に増加します。

現在、環境内の sink システムに ssn 値を保存している場合、またはスキーマレジストリーを使用している場合、この変更はシステムの動作に影響を与える可能性があります。

Debezium 2.3.4 では、"Autoset Replica Identity" と呼ばれる新しい PostgreSQL コネクター機能が導入されています。

PostgreSQL データベースは、レプリカ ID を使用して、挿入、更新、および削除イベントのデータベーストランザクションログにキャプチャーされた列を識別します。この機能を使用すると、テーブルのレプリカ ID 値を自動的に設定するようにコネクターを設定できます。コネクターが起動すると、レプリカ ID 設定が読み取られ、指定されたテーブルのレプリカ ID が設定されます。

新しい設定プロパティー plica.identity.autoset.values は、テーブルとレプリカの ID タプルのコンマ区切りリストを指定します。プロパティーがテーブルのレプリカ ID を指定すると、その値は既存のレプリカ ID 設定を上書きします。PostgreSQL レプリカアイデンティティータイプの詳細は、PostgreSQL のドキュメントを参照してください。

plica.identity.autoset.values プロパティーではコンマ区切りのリストを使用できます。各要素は、<fully-qualified-table-name>:<replica-identity> の形式を使用します。次の例は、FULL レプリカ ID を持つように 2 つのテーブル (table1 および table2) を設定する方法を示しています。

{ "replica.identity.autoset.values": "public.table1:FULL,public.table2:FULL" }

コネクターがデータベースにアクセスするときに使用するユーザーアカウントには、テーブルのレプリカ ID を設定する権限が必要です。アカウントに十分な権限がない場合、replica.identity.autoset.values を使用しようとすると失敗します。このプロパティーを使用してレプリカ ID を自動的に設定できない場合は、必要な権限を持つデータベースアカウントからテーブルのレプリカ ID を手動で設定する必要があります。

このリリースでは、新しい通知サブシステムが導入されています。これにより、Debezium は、増分スナップショットや従来のスナップショットなど、さまざまなコネクター操作のステータスを報告するイベントを発行できるようになります。この新しいサブシステムを使用すると、Kafka トピック、ログファイル、Java Management Extensions (JMX) など、いくつかの異なるチャネルを通じて通知を送信できます。これらの通知イベントは、さまざまな外部システムで使用できます。通知イベントは、以下のフィールドを含む一連のキー/値タプルとして表されます。

次の例は、単純な通知イベントを示しています。

{
  "id": "c485ccc3-16ff-47cc-b4e8-b56a57c3bad2",
  "aggregate_type": "Snapshot",
  "type": "Started",
  "additional_data": {
    ...
  }
}

{
  "id": "c485ccc3-16ff-47cc-b4e8-b56a57c3bad2",
  "aggregate_type": "Snapshot",
  "type": "Started",
  "additional_data": {
    ...
  }
}

本リリースでは、Debezium は以下のタイプの通知イベントをサポートします。

詳細は、コネクターのステータスをレポートするための通知設定 を参照してください。

このリリースでは、通知とチャネルのサブシステムが改善され、シグナルを通知に関連付けるようになりました。つまり、シグナルを送信して、Debezium によって消費されると、結果の通知には元のシグナルを参照する識別子が含まれるようになります。通信が複数のアプリケーションおよびプロセスに分散される場合、このメカニズムにより、プロセスは、結果として行われる操作とシグナルをこれまで以上に簡単に関連付けることができます。

Debezium は、リリース 1.7 で増分スナップショットが導入されて以来、シグナリングをサポートしています。シグナルは、メタデータを使用して、コネクターログへのエントリーの書き込みやアドホック増分スナップショットの実行などのタスクの実行を Debezium に指示するメカニズムを提供します。

このリリースでは、複数のシグナリングチャネルのサポートが導入され、Debezium がシグナルを監視し、シグナルに反応するために使用するメディアを指定できるようになりました。以前のバージョンでは、コネクター間で共通にサポートされるチャネルが 1 つあり、それがデータベースシグナルテーブルでした。本リリースでは、デフォルトで利用できます。

このリリースでは、シグナルチャネルのサブシステムが改善され、JMX 経由のシグナル送信をサポートするようになりました。JConsole ウィンドウには、コネクター用の 2 つのサブセクション、通知セクション、およびシグナルセクションが存在します。

signal セクションでは、JMX Bean で操作を呼び出して、シグナルを Debezium に送信できます。このシグナルは、3 つのパラメーターを受け入れるという点で、論理信号テーブル構造に似ています。

詳細は、Integration コネクターへのシグナルの送信 を参照してください。

Oracle Real Application Clusters (RAC) デプロイメントで Debezium Oracle コネクターを使用する場合は、rac.nodes 設定プロパティーを指定する必要があります。少なくとも、rac.nodes プロパティーでは、クラスター内の個々のノードのホストまたは IP アドレスを指定する必要があります。コネクターの古いバージョンでは、異なるノードが異なるポートを使用する可能性がある点を認識して、各ノードに一意のポート番号を指定できる別の形式もサポートしていました。

Debezium 2.3.4 では、Oracle RAC サポートが強化され、各ノードが異なる Oracle サイト識別子 (SID) を使用する可能性があることも認識されています。Oracle SID 設定のバリエーションを考慮して、rac.nodes 設定プロパティーで SID パラメーターを指定できるようになりました。

次の例は、それぞれ異なるポートと SID パラメーターを使用する 2 つの Oracle RAC ノードへの接続を示しています。

{ "connector.class": "io.debezium.connector.oracle.OracleConnector", "rac.nodes": "host1.domain.com:1521/ORCLSID1,host2.domain.com:1522/ORCLSID2", …​ }

Oracle は、OffsetScn、CurrentScn、OldestScn、CommittedScn など、JMX メトリックでさまざまなシステム変更番号 (SCN) 値を追跡します。これらの SCN 値は数値であり、LONG データ型の上限を超える場合があります。以前のリリースでは、Debezium は SCN 値を 文字列 値として公開していました。

これらのメトリックの有用性を向上させるために、Debezium はこれらの JMX メトリックを String 値ではなく BigInteger 値として公開するようになりました。この変更により、ユーザーは文字列ベースの値をサポートしていない Grafana や Prometheus などのツールを使用してこれらのメトリックの値を表示できるようになります。

データベースからエントリーを取得する場合、MongoDB および Oracle コネクターは、コネクター設定に設定された include および exclude フィルターをデータベースに送信できるようになりました。MongoDB コネクターはこれを自動的に行います。Oracle コネクターがエントリーの取得する時にフィルターを送信するようにするには、log.mining.query.filter.mode プロパティーをデフォルトの none 以外の値に設定します。

これまでのリリースでは、MongoDB コネクターと Oracle コネクターは最初にデータベースからイベントを取得し、次にフィルター設定に対してイベントを評価しました。このプロセスにより、データベースからネットワーク上のコネクターまでのすべての変更が効率的にシリアル化されます。特に大容量環境ではこのアプローチは非効率的です。コネクターは一部のイベントを受信しましたが、フィルター設定があるため、直後にそれらのイベントを破棄しました。クラウド環境で実行されるコネクターの場合、このように大量の過剰なデータを送信すると、使用コストが膨らみます。

コネクターが取得するデータの量を減らすために、Debezium 2.3.4 では、コネクターはデータの取得後にフィルターを評価しなくなりました。代わりに、包含リストと除外リストは、MongoDB 変更ストリームサブスクリプションまたは Oracle フェッチクエリーで定義されます。新しいアプローチでは、コネクターが読み取るイベントの数を減らすことで、ネットワークと CPU の使用率が低下します。完全なドキュメントまたはプレイメージ設定を使用して設定されたコネクターの場合は、ネットワークの使用率がさらに増えますが、これは全く不必要です。さらに、コネクターが処理を必要とするイベントのみを受信できるようにすることで、コネクターはより多くの処理を完了できるようになり、CPU 使用率が向上します。

以前のリリースでは、コネクターは起動時にフェイルファストストラテジーを使用していました。つまり、コネクターが起動ルーチンの完了に必要な手順 (データベースへの接続や認証など) を実行できない場合、コネクターは FAILED 状態になります。

状況によっては、コネクターが正常に開始され、一定期間実行された後、最終的に致命的なエラーが発生することがあります。エラーは、コネクターの起動ライフサイクル中にアクセスされなかったリソースに関連している可能性があるため、エラーの発生なしにコネクターを再起動できる場合があります。ただし、データベースが利用できなくなったことが原因で障害が発生した場合、コネクターの再起動後にデータベースが使用できなくなったままになると、コネクターが FAILED 状態になります。この場合、問題を解決するには人間が介入する必要があります。

信頼性と復元力を向上させるために、このリリースでは、コネクターは、起動時に使用できない可能性のあるリソースへのアクセスを試行するのではなく、ライフサイクルの後半でこれらのリソースへのアクセスを試行するようになりました。実質的には、Debezium は起動中は、利用できない可能性のあるリソースにアクセスすることに関しては厳密ではないので、Kafka Connect の再試行バックオフフレームワークを活用できます。

現在、コネクターの起動中にデータベースが利用できない場合でも、Kafka Connect の再試行が有効になっている限り、コネクターは失敗したリクエストを再試行し続けます。FAILED 状態は、再試行の最大回数に達した場合、または再試行不可能なエラーが発生した場合にのみ発生します。

Debezium の増分スナップショット機能は、再開可能で一貫性のあるデータのスナップショットを実行するメカニズムを提供します。スナップショットを再開するこの機能は、大量のデータを取り込む必要があるコネクターにとって重要です。

以前のリリースでは、増分スナップショットでは、スナップショットに含まれるすべてのテーブルにプライマリーキーを設定する必要がありました。Debezium 2.3.4 以降では、テーブルに "代理キー" として機能する一意のものが 1 つ含まれている限り、キーのないテーブルで増分スナップショットを実行できるようになりました。

インクリメンタルスナップショットシグナルで、代理キー列データを提供するには、シグナルペイロードに新しい代理キー属性 surrogate-key を含める必要があります。

{
  "data-collections": [ "public.mytab" ],
  "surrogate-key": "customer_ref"
}

{
  "data-collections": [ "public.mytab" ],
  "surrogate-key": "customer_ref"
}

前の例のシグナルは、public.mytab テーブルの増分スナップショットを開始します。スナップショットは、スナップショットウィンドウを生成するためのプライマリーキーとして customer_ref 列を使用します。

プライマリーキーが含まれるテーブルで代理キー機能を使用することもできます。たとえば、テーブルのプライマリーキーが複数の列で構成されている場合、代理キーには利点があります。複数の列に基づくクエリーでは、プライマリーキーの列ごとに論理和述語が生成され、パフォーマンスは環境に大きく依存する可能性があります。代理キーを使用してクエリー内の列の数を減らすと、より均一なパフォーマンスが得られます。

代理キーを使用すると、プライマリーキー列が文字ベースのデータ型をベースにするテーブルにも利点があります。一般にリレーショナルデータベースは、数値比較を行う場合と文字比較を行う場合の方が効率的であるため、数値代理キーを指定してクエリーのパフォーマンスを向上させることができます。

このリリースでは、イベントレコード変更 (ExtractChangedRecordState) の単一メッセージ変換 (SMT) が導入されています。この変換を使用すると、Debezium イベントのフィールドで、データベースの操作の後に値が変更されたもの、または変更されていないものを識別できます。変換を使用するには、以下のようにコネクター設定の一部としてその変換を設定します。

transforms=changes
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.changed=ChangedFields
transforms.changes.header.unchanged=UnchangedFields

transforms=changes
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.changed=ChangedFields
transforms.changes.header.unchanged=UnchangedFields

この変換に対して次のオプションを設定して、さまざまな種類の変更を示すことができます。

前の例と同様に、これらのオプションの両方を設定して、変更されたフィールドと変更されていないフィールドの両方を個別に表示できます。

変換を行うことで、指定された名前 (ChangedFields など) の新しいヘッダーが追加されます。次に、変更されたフィールドまたは変更されていないフィールドの名前を含むリストに、ヘッダー値を設定します。

ExtractChangedRecordState SMT の使用に関する詳細は、Debezium ユーザーガイドの イベントレコードの変更 を参照してください。

ExtractNewRecordState シングルメッセージ変換 (SMT) を使用して、Debezium 変更イベントを sink コネクターで使用できる簡素化された形式に変換できます。

このリリースでは、イベントのペイロードまたはメッセージキーからフィールドを削除するために使用できる、変換用の 3 つの新しい設定オプションが追加されました。

transforms=changes,extract
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.unchanged=UnchangedFields
transforms.extract.type=io.debezium.transforms.ExtractNewRecordState
transforms.extract.drop.fields.header.name=UnchangedFields

transforms=changes,extract
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.unchanged=UnchangedFields
transforms.extract.type=io.debezium.transforms.ExtractNewRecordState
transforms.extract.drop.fields.header.name=UnchangedFields

前述の設定では変更されていないフィールドがリストされますが、このようなフィールドはイベントペイロードから削除されません。指定したキーのフィールドが変更されなかった場合、設定は drop.fields.from.key のデフォルトの false 値を明示的に変更しないため、そのフィールドは保持されます。

SMT によってイベントペイロード内の必須フィールドが変更されなかったことが原因でドロップされる場合、スキーマ互換性を確保するために、フィールドは出力内に保持されます。

ExtractNewRecordState SMT の詳細は、Debezium 変更イベントからの after 状態のソースレコードの抽出 を参照してください。

イベントレコードから指定されたヘッダーフィールドを抽出し、そのヘッダーフィールドをイベントレコード内の値にコピーまたは移動します。詳細は、Debezium ユーザーガイド メッセージヘッダーのイベントレコード値への変換 を参照してください。

PartitionRouting SMT を使用すると、1 つ以上の指定されたペイロードフィールドの値に基づいて、イベントを特定の宛先パーティションにルーティングできます。Debezium は、宛先パーティションを計算するために、指定したフィールド値のハッシュを生成します。

詳細は、Debezium ユーザーガイドの ペイロードフィールドに基づくレコードのパーティションへのルーティング を参照してください。

過去のリリースでは、シャードクラスターデプロイメントで Debezium MongoDB コネクターを使用すると、コネクターにより、シャード内の各レプリカセットとの直接接続が確立されました。このアプローチは、コネクターが mongos ルーターインスタンスとの接続を開く必要がある という MongoDB 提案と競合します。

このリリースでは、推奨される接続ストラテジーを使用するようにコネクターが再設計されました。シャードクラスターでコネクターを使用する場合は、コネクターが mongos インスタンスに接続するように設定を調整します。その他の変更は必要ありません。

MongoDB マルチレプリカおよびシャードクラスターで増分スナップショットを使用できるようになりました。詳細は、{NameUserGuide} の MongoDB の章の増分スナップショット を参照してください。

このリリースでは、リレーショナルデータベースおよび非リレーショナルデータベース用のソースコネクターのみを長年の間開発してきたことを打ち切り、Debezium は JDBC sink コネクターの実装を導入しました。Debezium JDBC sink コネクターは、最初にイベントフラット化変換を適用せずに Debezium コネクターによって出力される未加工の変更イベントを取得できるという点で、他のベンダーの実装とは異なります。Debezium JDBC sink コネクターは、列型の伝播などのネイティブの Debezium ソースコネクター機能を利用できるため、データパイプラインの処理フットプリントを削減し、設定を簡素化できる可能性があります。

次の例は、orders と呼ばれる Kafka トピックから変更イベントを PostgreSQL データベースに取り込むための簡単な設定を示しています。トピック内のイベントは、ExtractNewRecordState 変換を使用せずに、Debezium MySQL コネクターによって発行されています。

{
  "name": "mysql-to-postgres-pipeline",
  "config": {
    "connector_class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc://postgresql://<host>:<port>/<database>",
    "connection.user": "<username>",
    "connection.password": "<password>",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "primary.key.mode": "record_key",
    "schema.evolution": "basic"
  }
}

{
  "name": "mysql-to-postgres-pipeline",
  "config": {
    "connector_class": "io.debezium.connector.jdbc.JdbcSinkConnector",
    "topics": "orders",
    "connection.url": "jdbc://postgresql://<host>:<port>/<database>",
    "connection.user": "<username>",
    "connection.password": "<password>",
    "insert.mode": "upsert",
    "delete.enabled": "true",
    "primary.key.mode": "record_key",
    "schema.evolution": "basic"
  }
}

前述の例は、宛先 PostgreSQL データベースにアクセスするための接続文字列と認証情報を定義する一連の connection.* プロパティーを示しています。宛先データベースに書き込む場合、レコードは UPSERT セマンティクスを使用し、レコードが存在しない場合は挿入を使用してレコードを作成し、レコードが存在する場合はレコードを更新します。スキーマ進化が有効になり、テーブルのキー列がイベントのプライマリーキーから派生します。

以下のリレーショナルデータベースでこのリリースの JDBC sink コネクターを使用できます。

詳細は、JDBC の Debezium コネクター を参照してください。

Debezium の初期スナップショットプロセスは、リレーショナルデータベースでは常にシングルスレッドでした。この制限は主に、複数のトランザクション間でデータの一貫性を確保するのが複雑であることに起因します。

このリリース以降、一貫性のあるデータベーススナップショットを実行するときに複数のスレッドを使用するようにコネクターを設定できるようになりました。この実装では、これらの複数のスレッドを使用し、テーブルレベルのスナップショットを並行して実行します。

並列スナップショットを利用するには、コネクター設定で snapshot.max.threads プロパティーを指定し、それに 1 より大きい値を割り当てます。

snapshot.max.threads=4

snapshot.max.threads=4

前の例に基づいて、コネクターは最大 4 つのテーブルのスナップショットを並行して作成します。スナップショットを作成するテーブルがさらにある場合、1 つのスレッドが終了した後、コネクターはキュー内の次のテーブルを処理します。このプロセスは、すべてのテーブルのスナップショットが作成されるまで続行されます。

Oracle 用の Debezium コネクターは、Oracle ログライターバッファー (LGWR) プロセスのフラッシュサイクルを監視するための内部 フラッシュテーブル を管理します。コネクターがデータベースへのアクセス時に使用するユーザーアカウントには、フラッシュテーブルの作成と書き込みの権限が必要です。ロジカルスタンバイデータベースにはデータ操作に関してより制限的なルールがあり、読み取り専用である場合もあるため、データベースへの書き込みは好ましくないことや、許可されないことさえあります。

コネクターが Oracle の読み取り専用ロジカルスタンバイデータベースから変更を取り込めるようにするために、このリリースでは、この フラッシュテーブル の作成と管理を無効にするフラグが導入されています。この開発者プレビュー機能は、Oracle スタンドアロンおよび Oracle RAC インストールの両方で使用できます。

コネクターが Oracle 読み取り専用ロジカルスタンバイから変更を取り込めるようにするには、次のコネクターオプションを追加します。

internal.log.mining.read.only=true

internal.log.mining.read.only=true

Debezium は従来、最低 1 回は行うデリバリーソリューションで、変更が見逃されることはありませんでした。1 回限りのデリバリーは、KIP-618 の一部として Apache Kafka コミュニティーによって提案されています。この提案は、プロデューサー (ソースコネクター) が再試行中に遭遇する一般的な問題に対処することを目的としています。ブローカーがすでにバッチをコミットしている場合でも、コネクターはイベントのバッチを Kafka ブローカーに再送信する場合があります。この状況では、重複したイベントが送信される可能性があり、重複を簡単に処理できないコンシューマー (sink コネクター) に問題が発生する可能性があります。

1 回限りのデリバリーを利用するためにコネクター設定を変更する必要はありません。ただし、1 回限りのデリバリーを有効にするには、KIP-618 で導入された設定プロパティー を使用するように Kafka Connect ワーカー設定を調整する必要があります。Debezium 2.3.4 では、PostgreSQL の 1 回限りのセマンティクスはスナップショット中ではなく、ストリーミングフェーズ中にのみ適用されます。