# Amazon MSK イベントソース向け Lambda イベントソースマッピング作成
<a name="msk-esm-create"></a>

イベントソースマッピングの作成には、Lambda コンソール、[AWS Command Line Interface (CLI)](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html)、[AWS SDK](https://aws.amazon.com/getting-started/tools-sdks/) のいずれかを使用できます。

**注記**  
イベントソースマッピングを作成すると、Lambda は MSK クラスターを含むプライベートサブネットに[ハイパープレーン ENI](configuration-vpc.md#configuration-vpc-enis) を作成し、これにより Lambda からの安全な接続を確立します。このハイパープレーン ENI は Lambda 関数ではなく、MSK クラスターのサブネットおよびセキュリティグループ設定を使用します。

次のコンソールの手順で、Lambda 関数のトリガーとして Amazon MSK クラスターを追加します。内部でイベントソースマッピングリソースが作成されます。

**Amazon MSK トリガーを Lambda 関数 (コンソール) に追加するには**

1. Lambda コンソールの [[関数]](https://console.aws.amazon.com/lambda/home#/functions) ページを開きます。

1. Amazon MSK トリガーを追加する Lambda 関数の名前を選択します。

1. **[関数の概要]** で **[トリガーを追加]** をクリックします。

1. **[トリガー設定]** で **[MSK]** を選択します。

1. Kafka クラスターの詳細を指定するには、次の手順を実行してください。

   1. [**MSK cluster**] (MSK クラスター) で、クラスターを選択します。

   1. **[トピック名]**に、メッセージを使用する Kafka トピックの名前を入力します。

   1. **[コンシューマーグループ ID]** に、参加する Kafka コンシューマーグループの ID を入力します (該当する場合)。詳細については、「[Lambda でのカスタマイズ可能なコンシューマーグループ ID](kafka-consumer-group-id.md)」を参照してください。

1. **[クラスター認証]** で必要な設定を行います。クラスター認証の詳細については、「[Lambda での Amazon MSK クラスターの認証方法の設定](msk-cluster-auth.md)」を参照してください。
   + 接続の確立時に Lambda で MSK クラスターに対する認証を実行させたい場合、**[認証の使用]** をオンに切り替えます。認証の使用をお勧めします。
   + 認証を使用する場合、**[認証方法]** で使用する認証方法を選択します。
   + 認証を使用する場合、**[Secrets Manager キー]** で、クラスターへのアクセスに必要な認証情報を含む Secrets Manager キーを選択します。

1. **[イベントポーラー設定]** で、必要な設定を行います。
   + **[トリガーを有効にする]** を選択し、作成直後にトリガーが有効化されるようにします。
   + イベントソースマッピングに **[プロビジョニングモードの設定]** を行うかどうかを選択します。詳細については、「[Lambda での Apache Kafka イベントポーラーのスケーリングモード](kafka-scaling-modes.md)」を参照してください。
     + プロビジョンドモードを設定する場合は、**[最小イベントポーラー]** の値、**[最大イベントポーラー]** の値、PollerGroupName のオプションの値を入力して、同じイベントソース VPC 内の複数の ESM のグループ化を指定します。
   + **[開始位置]** で、Lambda がストリームからの読み取りを開始する方法を選択します。詳細については、「[Lambda での Apache Kafka ポーリングとストリームの開始位置](kafka-starting-positions.md)」を参照してください。

1. **[バッチ処理]** で必要な設定を行います。バッチ処理の詳細については、「[バッチ処理動作](invocation-eventsourcemapping.md#invocation-eventsourcemapping-batching)」を参照してください。

   1. [**Batch size**] (バッチサイズ) で、単一バッチで取得されるメッセージの最大数を設定します。

   1. **[バッチウィンドウ]** に、Lambda が関数呼び出し前にレコード収集に費やすことのできる最大時間 (秒) を入力します。

1. **[フィルタリング]**で必要な設定を行います。のフィルタリングについての詳細は「[Amazon MSK およびセルフマネージド Apache Kafka イベントソースからのイベントのフィルタリング](kafka-filtering.md)」を参照してください。
   + **[フィルター条件]** にフィルター条件定義を追加して、イベントを処理するかどうかを決定します。

1. **[障害処理]** で必要な設定を行います。障害処理の詳細については、「[Amazon MSK とセルフマネージド Apache Kafka イベントソースの破棄されたバッチのキャプチャ](kafka-on-failure.md)」を参照してください。
   + **[障害発生時の宛先]** には、障害発生時の送信先の ARN を指定します。

1. **[タグ]** には、このイベントソースマッピングに関連付けるタグを入力します。

1. トリガーを追加するには、[**Add**] (追加) を選択します。

[create-event-source-mapping](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) コマンドと一緒に AWS CLI を使用してイベントソースマッピングを作成することもできます。次の例では、`LATEST` メッセージを開始点として Lambda 関数の `my-msk-function` を `AWSKafkaTopic` トピックにマッピングするイベントソースマッピングを作成します。このコマンドは [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) オブジェクトも使用して、クラスターに接続するときに [SASL/SCRAM](msk-cluster-auth.md#msk-sasl-scram) 認証を使用するように Lambda に指示します。

```
aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "SASL_SCRAM_512_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
```

クラスターが [mTLS 認証](msk-cluster-auth.md#msk-mtls)を使用する場合、`CLIENT_CERTIFICATE_TLS_AUTH` および Secrets Manager のキー ARN を指定する [SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) オブジェクトを含める必要があります。これは次のコマンドで示されます。

```
aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
  --source-access-configurations '[{"Type": "CLIENT_CERTIFICATE_TLS_AUTH","URI": "arn:aws:secretsmanager:us-east-1:111122223333:secret:my-secret"}]'
```

クラスターが [IAM 認証](msk-cluster-auth.md#msk-iam-auth)を使用する場合、[SourceAccessConfiguration](https://docs.aws.amazon.com/lambda/latest/api/API_SourceAccessConfiguration.html) オブジェクトは不要です。これは次のコマンドで示されます。

```
aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:cluster/my-cluster/fc2f5bdf-fd1b-45ad-85dd-15b4a5a6247e-2 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function
```