Amazon MSK で EventBridge Pipes をデプロイする方法を教えてください。

解決策

注: Amazon MSK の詳細とクラスターの作成方法については、「Amazon MSK の使用を開始する」を参照してください。

前提条件

• MSK クラスター認証 EventBridge には、MSK クラスターへのアクセス、レコードの取得、その他のタスクの実行用のアクセス許可が必要です。
• ネットワーク設定 EventBridge は、MSK クラスターに関連付けられた Amazon 仮想プライベートクラウド (Amazon VPC) リソースにアクセスできる必要があります。
• Amazon MSK 実行ロールのアクセス許可 EventBridge には、MSK トピックに関連するリソースを管理するためのアクセス許可が必要です。

Amazon MSK で EventBridge Pipes をデプロイする

次の手順を実行します。

1. 対応する AWS リージョンで EventBridge コンソールに移動します。
2. ナビゲーションパネルで [パイプ] を選択します。
3. [パイプを作成] を選択します。
4. パイプ名を入力します。
5. (オプション) パイプの説明を入力します。

パイプのソースとして Amazon MSK を選択します

次の手順を実行します。

1. [ソースを選択] で [Amazon MSK] を選択します。
2. 作成した Amazon MSK クラスターを選択します。
3. Apache Kafka クラスターにレコードを保存するために使用する Apache Kafka トピックの名前を入力します。
4. (オプション) 参加する Apache Kafka コンシューマーグループの ID を指定します。詳細については、「EventBridge Pipes のソースに使用する Amazon Managed Streaming for Apache Kafka」を参照してください。
5. (オプション) 認証方法とシークレットキーを選択します。

追加設定を適用する

次の手順を実行します。

1. (オプション) [バッチサイズ] と [バッチウィンドウ] を設定します。
2. 開始位置を設定します。
   注: 開始位置のデフォルト設定は、latest です。必要に応じて、Trim horizon に変更します。 Trim horizon は、Apache Kafka における Earliest と同等です。
3. (オプション) [フィルター] と [エンリッチメント] の設定を適用します。
4. [ターゲットサービス] リストからターゲットを選択します。
5. (オプション) 入力トランスフォーマーを定義します。
6. パイプを正しく構成したことを確認します。次に、[パイプを作成] を選択します。

パイプが Running ステータスに移行すると、EventBridge はソースからの新しいメッセージを内部でポーリングします。次に、ターゲットを同期的に呼び出します。

注: イベントのペイロードにはメッセージの配列が含まれます。各項目には、タイムスタンプと base64 でエンコードされたメッセージとともに、MSK トピックと MSK パーティション識別子の詳細が含まれています。

MSK イベントの例を次に示します。

[  {
    "eventSource": "aws:kafka",
    "eventSourceArn": "arn:aws:kafka:sa-east-1:123456789012:cluster/vpc-2priv-2pub/751d2973-a626-431c-9d4e-d7975eb44dd7-2",
    "eventSourceKey": "mytopic-0",
    "topic": "mytopic",
    "partition": "0",
    "offset": 15,
    "timestamp": 1545084650987,
    "timestampType": "CREATE_TIME",
    "key": "abcDEFghiJKLmnoPQRstuVWXyz1234==",
    "value": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
    "headers": [
      {
        "headerKey": [
          104,
          101,
          97,
          100,
          101,
          114,
          86,
          97,
          108,
          117,
          101
        ]
      }
    ]
  }
]

呼び出しに関連するエラーのトラブルシューティング

CloudWatch コンソールで該当するリージョンの Amazon CloudWatch メトリクスを参照し、有効化、呼び出し、スロットリングに関連する問題を切り分けます。

パイプを作成、削除、更新する際、操作がエラー状態になることがあります。詳細については、「パイプのエラー状態」を参照してください。

パイプの呼び出し時には、主にパイプの内部エラーと顧客呼び出しでのエラーの 2 種類のエラーが発生する可能性があります。

顧客のターゲットサービス呼び出しを試行した際に HTTP 接続が失敗すると、パイプの内部エラーが発生することがあります。パイプサービス自体の可用性が一時的に低下した場合も、エラーが発生する可能性があります。

パイプへのアクセス許可が不十分でターゲットの呼び出しに失敗した場合、顧客呼び出しでエラーが発生する可能性があります。同期的に呼び出した、顧客の AWS Lambda、AWS Step Functions、API 送信先、または Amazon API Gateway エンドポイントにロジックエラーがある場合も、エラーが発生する可能性があります。詳細については、「アクセス許可」を参照してください。

EventBridge Pipes は、別のアカウントからの Amazon MQ ブローカーのクロスアカウント処理をサポートしていません。詳細については、「Amazon EventBridge Pipes のエラー処理とトラブルシューティング」を参照してください。

関連情報

Amazon EventBridge パイプを作成する