Application Integration でサポートされているコネクタをご覧ください。

Apache Kafka トリガー

Apache Kafka トリガーは、Apache Kafka 接続に登録している Apache Kafka イベントに基づいて統合を呼び出すことができるコネクタイベントトリガーです。

始める前に

Apache Kafka トリガーの新しい接続を作成または構成する場合は、プロジェクトに対する次の Identity and Access Management（IAM）ロールがあることを確認してください。

Connectors 管理者（ roles/connectors.admin ）

ロールの付与については、アクセスの管理をご覧ください。

Apache Kafka トリガーに使用するサービスアカウントに、次の IAM ロールを付与します。

Application Integration の起動元（ roles/integrations.integrationInvoker ）

サービスアカウントへのロールの付与については、サービスアカウントに対するアクセス権の管理をご覧ください。

制約事項

Apache Kafka トリガーには次の制限があります。

イベントサブスクリプションで SASL_SSL セキュリティプロトコルをサポートします。
イベントサブスクリプションで mTLS はサポートされていません。

Apache Kafka トリガーを追加する

統合に Apache Kafka トリガーを追加する手順は次のとおりです。

Google Cloud コンソールで [Application Integration] ページに移動します。
Application Integration に移動
ナビゲーションメニューで [統合] をクリックします。
[統合リスト] ページが開き、Google Cloud プロジェクトで使用可能なすべての統合が一覧表示されます。
既存の統合を選択するか、[統合の作成] をクリックして新しい統合を作成します。
新しい統合を作成する場合:
1. [統合の作成] ペインで名前と説明を入力します。
2. 統合のリージョンを選択します。
  注: [リージョン] プルダウンには、Google Cloud プロジェクトにプロビジョニングされたリージョンのみが表示されます。新しいリージョンをプロビジョニングするには、[リージョンを有効にする] をクリックします。詳細については、新しいリージョンを有効にするをご覧ください。
3. 統合用のサービスアカウントを選択します。統合のサービスアカウントの詳細は、統合ツールバーの [統合の概要] ペインでいつでも変更または更新できます。
  注: サービスアカウントを選択するオプションは、選択したリージョンで統合ガバナンスを有効にしている場合にのみ表示されます。
4. [作成] をクリックします。
統合エディタで統合が開きます。
統合エディタのナビゲーションバーで、[トリガー] をクリックして、使用可能なトリガーのリストを表示します。
統合エディタで、[Apache Kafka トリガー] 要素をクリックして配置します。

Apache Kafka トリガーを構成するには、Integration Connectors で使用可能な既存の Apache Kafka 接続を使用するか、インライン接続作成オプションを使用して新しい Apache Kafka 接続を作成します。

既存の接続を使用して Apache Kafka トリガーを構成する

Integration Connectors で既存の Apache Kafka 接続を使用して、Apache Kafka トリガーを構成できます。なお、Apache Kafka トリガーはコネクタイベントトリガーであるため、トリガーの構成にはイベントサブスクリプションを有効にした Apache Kafka 接続のみが使用できます。

新しい Apache Kafka 接続を使用して Apache Kafka トリガーを構成する方法については、新しい接続を使用して Apache Kafka トリガーを構成するを参照してください

既存の Apache Kafka 接続を使用して Apache Kafka トリガーを構成するには、次の手順を実行します。

統合エディタで [Apache Kafka トリガー] 要素をクリックして、トリガー構成ペインを開きます。
[トリガーの構成] をクリックします。
[Connector Event Trigger Editor] ページで、構成の詳細を次のとおりに入力します。
1. リージョン: Apache Kafka 接続のリージョンを選択します。
2. 接続: 使用する Apache Kafka 接続を選択します。
  Application Integration には、アクティブでイベントサブスクリプションが有効になっている Apache Kafka 接続のみが表示されます。
3. トピック名: トピック名を入力します。
4. コンシューマグループ ID: 必要に応じて、Kafka コンシューマが属するコンシューマグループの名前を指定するコンシューマグループ ID を入力します。
5. Partition：特定のパーティションにサブスクライブする場合は、パーティションを入力します。パーティションとは、トピックを複数の部分に分割し、各部分をクラスタ内の別のノードに保存する方法です。
6. 初期オフセット：オフセットリセット戦略を次のように選択します。
  - [最も早い] を選択すると、接続では、コンシューマグループの存続期間より前に生成されたメッセージを含む、未消費のメッセージが使用されます。
  - [最も新しい] を選択すると、コンシューマグループの作成後に生成されたメッセージが接続で使用されます。
7. サービスアカウント：Apache Kafka トリガーに必要な IAM ロールを持つサービスアカウントを選択します。
[完了] をクリックします。

新しい接続を使用して Apache Kafka トリガーを構成する

新しい Apache Kafka 接続を使用して Apache Kafka トリガーを構成するには、次の手順に従います。

統合エディタで [Apache Kafka trigger] 要素をクリックして、トリガー構成ペインを開きます。
Apache Kafka トリガーの構成をクリックします。
[リージョン] フィールドはスキップします。
[接続] をクリックし、リストから [接続の作成] オプションを選択します。[接続を作成] ペインが表示されます。
[ロケーション] ステップで、新しい Apache Kafka 接続のロケーションを次のように選択します。
1. リージョン：リストからリージョンを選択します。
2. [次へ] をクリックします。
[接続の詳細] ステップで、新しい Apache Kafka 接続の詳細を入力します。
1. コネクタのバージョン：リストから利用可能な Apache Kafka コネクタのバージョンを選択します。
2. 接続名: Apache Kafka 接続の名前を入力します。
  注: 接続名は次の条件を満たす必要があります。
  - 接続名には英字、数字、ハイフンを使用できます。
  - 文字は小文字のみを使用できます。
  - 接続名の先頭には英字を設定し、末尾には英字または数字を設定する必要があります。
  - 接続名は 63 文字以内で指定してください。
3. （省略可）説明文: 接続の説明を入力します。
4. （省略可）Cloud Logging を有効にする: 接続のすべてのログデータを保存するには、このチェックボックスをオンにします。
5. サービスアカウント: Apache Kafka 接続に必要な IAM ロールを持つサービスアカウントを選択します。
6. Apache Kafka 接続では、[イベントのサブスクリプション、エンティティ、アクションを有効にする] オプションがデフォルトで選択されています。
  注: このコネクタは Enable only event subscription オプションをサポートしていません。
7. 型検出スキーム: MessageOnly を選択します。
8. レジストリサービス: トピックスキーマの操作に使用される Schema Registry サービス。
9. レジストリタイプ: 特定のトピックに指定されたスキーマのタイプ。
10. レジストリバージョン: 指定したトピックの RegistryUrl から読み取られたスキーマのバージョン。
11. レジストリユーザー: RegistryUrl で指定されたサーバーを承認するユーザー名またはアクセスキーの値。
12. レジストリパスワード: RegistryUrl で指定されたサーバーで承認するパスワードと秘密鍵のキー値を含む Secret Manager シークレット。
13. 必要に応じて、接続ノードの設定を構成します。
  - ノードの最小数: 接続ノードの最小数を入力します。
  - ノードの最大数: 接続ノードの最大数を入力します。
  ノードは、トランザクションを処理する接続の単位（またはレプリカ）です。1 つの接続でより多くのトランザクションを処理するには、より多くのノードが必要になります。逆に、より少ないトランザクションを処理するには、より少ないノードが必要になります。ノードがコネクタの料金に与える影響については、接続ノードの料金をご覧ください。値を入力しない場合は、デフォルトで最小ノード数は 2 に設定され（可用性を高めるため）、最大ノード数は 50 に設定されます。
14. 必要に応じて、[+ ラベルを追加] をクリックしてキーと値ペアの形式でラベルを接続に追加します。
15. SSL を有効にする: このフィールドは SSL を有効にするかどうかを設定します。
16. [次へ] をクリックします。
[宛先] セクションに、接続するリモートホスト（バックエンドシステム）の詳細を入力します。
1. 宛先の種類: 宛先の種類を選択します。
  - リストからホストアドレスを選択して、宛先のホスト名または IP アドレスを指定します。
  - バックエンドシステムへのプライベート接続を確立する場合は、リストからエンドポイントアタッチメントを選択し、次にエンドポイントアタッチメントリストから必要なエンドポイントアタッチメントを選択します。
    注：エンドポイントアタッチメントを作成する方法については、PSC サービスアタッチメントとエンドポイントアタッチメントをご覧ください。エンドポイントアタッチメントを作成すると、Endpoint Attachment リストに表示されます。
  セキュリティをさらに強化してバックエンドシステムへのパブリック接続を確立する場合は、接続用の静的アウトバウンド IP アドレスの構成を検討してから、特定の静的 IP アドレスのみを許可リストに登録するようファイアウォールルールを構成します。
  
  他の宛先を入力するには、[+ 宛先を追加] をクリックします。
2. [次へ] をクリックします。
[認証] セクションで、認証の詳細を入力します。
1. [認証タイプ] を選択し、関連する詳細を入力します。
  Apache Kafka 接続でサポートされる認証タイプは次のとおりです。
  - ユーザー名とパスワード
    - ユーザー名: 接続に使用する Apache Cassandra ユーザー名。
    - パスワード: Apache Kafka ユーザー名に関連付けられているパスワードを含む Secret Manager のシークレット。
    - Auth Scheme: 認証に使用されるスキーム
      Apache Kafka 接続でサポートされる認証スキームは次のとおりです。
      - Plain
      - SCRAM-SHA-1
      - SCRAM-SHA-256
  - 利用不可
    匿名ログインを使用する場合は、[Not Available] を選択します。
2. [次へ] をクリックします。
デッドレター構成を入力します。デッドレターを構成すると、指定された Pub/Sub トピックに未処理のイベントが書き込まれます。次の詳細情報を入力します。
1. デッドレタープロジェクト ID: デッドレター Pub/Sub トピックを構成した Google Cloud プロジェクト ID。
2. デッドレタートピック: 未処理イベントの詳細を書き込む Pub/Sub トピック。
[次へ] をクリックします。
Review: 接続と認証の詳細を確認します。
[作成] をクリックします。

トリガー出力

Apache Kafka トリガーがイベントサブスクリプションの構成を完了するまでに数分かかります。イベントサブスクリプションのステータスは、トリガー構成ペインの [Event Subscription details] で確認できます。

Apache Kafka トリガーは、次の状態を使用してイベントサブスクリプションのステータスを示します。

Creating: トリガーがイベントサブスクリプションに登録中であることを示します。
Active: トリガーがイベントサブスクリプションに正常に登録されたことを示します。
Error: 構成されたイベントサブスクリプションに問題があることを示します。

[Event Subscription details] セクションには、イベントサブスクリプションのステータスに加えて、接続リージョン、接続名、イベントサブスクリプション名の詳細が表示されます。

トリガー出力変数

イベントごとに、Apache Kafka トリガーは、ダウンストリームタスクで使用できる ConnectorEventPayload 出力変数（JSON 形式）を生成します。出力変数には、Apache Kafka イベントからのペイロードが含まれます。次に例を示します。

  {
    "message": "{\"key\":\"18\",\"topic\":\"test\",\"value\":\"hello world\",\"partition\":0,\"offset\":12,\"headers\":\"{}\",\"timestamp\":1712755076824,\"serializedMessagesize\":14}",
    "contentType": "text/plain"
 }

イベントサブスクリプションの表示

Integration Connectors で接続に関連付けられているすべてのイベントサブスクリプションの表示と管理を行う手順は、次のとおりです。

[Integration Connectors] > [接続] ページに移動します。
[接続] ページに移動
サブスクリプションを表示する接続をクリックします。
[イベントサブスクリプション] タブをクリックします。
ここでは、接続のすべてのサブスクリプションが表示されます。

Apache Kafka トリガーを編集

Apache Kafka トリガーを編集して、接続構成とイベントサブスクリプションの詳細を変更または更新できます。

Apache Kafka トリガーを編集するには、次の操作を行います。

統合エディタで [Apache Kafka trigger] 要素をクリックして、トリガー構成ペインを開きます。
Apache Kafka トリガーの構成をクリックします。
[Connector Event Trigger Editor] ページで、次の操作を行います。
1. 構成済みのイベントサブスクリプションを保持するには、[保持] をクリックします。それ以外の場合は、[削除] をクリックします。
  警告: イベントサブスクリプションを削除すると、そのイベントサブスクリプションを使用して公開されているすべての統合に影響します。このタスクは取り消すことができません。
2. 接続の構成とイベントサブスクリプションの詳細を必要に応じて更新します。
3. [完了] をクリックします。

更新された接続とイベントサブスクリプションの詳細は、トリガー構成ペインの [Event Subscription details] で確認できます。