Debezium を使用して Spanner Omni に接続する

Debezium は、変更データ キャプチャ（CDC）用のオープンソース分散プラットフォームです。Spanner 用の Debezium コネクタは、Spanner 変更ストリームから行レベルの変更をキャプチャし、Apache Kafka トピックにストリーミングします。

Debezium コネクタは、Spanner で動作するのと同じように Spanner Omni で動作します。このドキュメントでは、Spanner Omni に接続するように Debezium コネクタを構成する方法について説明します。

Spanner Omni Debezium 接続は、プレーン テキスト、TLS、相互 TLS（mTLS）の 3 つのセキュリティ構成をサポートしています。

始める前に

Spanner Omni で Debezium コネクタを使用するには、次の前提条件を満たしていることを確認してください。

  • Kafka Connect プラグイン ディレクトリに Debezium connector for Spanner プラグイン バージョン 3.7.0 以降をダウンロードしてインストールします。
  • Apache Kafka と Kafka Connect を設定します。詳細については、Debezium インストール ガイドをご覧ください。
  • Spanner Omni でデータベースと変更ストリームを作成します。詳細については、Spanner ドキュメントの変更ストリームをご覧ください。

Debezium コネクタを構成する

Debezium コネクタを Spanner Omni に接続するには、Kafka Connect コネクタ構成で次のプロパティを指定します。

  • connector.class: io.debezium.connector.spanner.SpannerConnector に設定します。
  • spanner.type: OMNI に設定します。OMNI に設定すると、gcp.spanner.project.id と gcp.spanner.instance.id が自動的に default に設定され、構成で指定する必要はありません。
  • gcp.spanner.host: Spanner Omni エンドポイント。たとえば、プレーン テキストの場合は http://localhost:15000、TLS と mTLS の場合は https://localhost:15000。
  • gcp.spanner.database.id: Spanner Omni データベースの ID。
  • gcp.spanner.change-stream.name: キャプチャする変更ストリームの名前。
  • spanner.omni.use.plaintext: 省略可。プレーン テキスト接続を確立するには、true に設定します。
  • spanner.omni.client.cert.path: 省略可。mTLS 接続のクライアント証明書ファイルのパス。
  • spanner.omni.client.key.path: 省略可。mTLS 接続用の PKCS#8 形式のクライアント秘密鍵ファイルへのパス。

接続を確立する

次の例は、サポートされている各セキュリティ構成に対して Debezium コネクタを構成する方法を示しています。

書式なしテキスト

プレーン テキスト接続を確立するには、spanner.type を OMNI に設定し、http:// でエンドポイントを指定して、spanner.omni.use.plaintext を true に設定します。

{
  "name": "CONNECTOR_NAME",
  "config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "1",
    "spanner.type": "OMNI",
    "gcp.spanner.host": "http://ENDPOINT",
    "gcp.spanner.database.id": "DATABASE_ID",
    "gcp.spanner.change-stream.name": "CHANGE_STREAM_NAME",
    "spanner.omni.use.plaintext": "true"
  }
}

次のように置き換えます。

  • CONNECTOR_NAME: Debezium コネクタ インスタンスの名前（例: spanner-omni-connector）。

  • ENDPOINT: Spanner Omni インスタンスのエンドポイント（例: localhost:15000）。

  • DATABASE_ID: Spanner Omni データベースの ID（例: test-db）。

  • CHANGE_STREAM_NAME: データベース内の変更ストリームの名前（例: my_change_stream）。

TLS

TLS 接続を確立するには、Java トラストストアを構成するで説明されているように、Spanner Omni CA 証明書を Java トラストストアに追加します。spanner.type を OMNI に設定し、https:// を使用してエンドポイントを指定します。

{
  "name": "CONNECTOR_NAME",
  "config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "1",
    "spanner.type": "OMNI",
    "gcp.spanner.host": "https://ENDPOINT",
    "gcp.spanner.database.id": "DATABASE_ID",
    "gcp.spanner.change-stream.name": "CHANGE_STREAM_NAME"
  }
}

次のように置き換えます。

  • CONNECTOR_NAME: Debezium コネクタ インスタンスの名前（例: spanner-omni-connector）。

  • ENDPOINT: Spanner Omni インスタンスのエンドポイント（例: localhost:15000）。

  • DATABASE_ID: Spanner Omni データベースの ID（例: test-db）。

  • CHANGE_STREAM_NAME: データベース内の変更ストリームの名前（例: my_change_stream）。

mTLS

mTLS 接続を確立するには、Java トラストストアを構成するで説明されているように、Spanner Omni CA 証明書を Java トラストストアに追加します。spanner.type を OMNI に設定し、https:// を使用してエンドポイントを指定し、クライアント証明書とクライアント秘密鍵のパスを指定します。クライアントの秘密鍵は、Java SDK の mTLS の手順で説明されているように、Java 準拠の PKCS#8 形式である必要があります。

{
  "name": "CONNECTOR_NAME",
  "config": {
    "connector.class": "io.debezium.connector.spanner.SpannerConnector",
    "tasks.max": "1",
    "spanner.type": "OMNI",
    "gcp.spanner.host": "https://ENDPOINT",
    "gcp.spanner.database.id": "DATABASE_ID",
    "gcp.spanner.change-stream.name": "CHANGE_STREAM_NAME",
    "spanner.omni.client.cert.path": "PATH_TO_CLIENT_CERT",
    "spanner.omni.client.key.path": "PATH_TO_CLIENT_KEY"
  }
}

次のように置き換えます。

  • CONNECTOR_NAME: Debezium コネクタ インスタンスの名前（例: spanner-omni-connector）。

  • ENDPOINT: Spanner Omni インスタンスのエンドポイント（例: localhost:15000）。

  • DATABASE_ID: Spanner Omni データベースの ID（例: test-db）。

  • CHANGE_STREAM_NAME: データベース内の変更ストリームの名前（例: my_change_stream）。

  • PATH_TO_CLIENT_CERT: クライアント証明書ファイルのパス。

  • PATH_TO_CLIENT_KEY: PKCS#8 形式のクライアント秘密鍵ファイルのパス。

次のステップ