翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# MSK
`MSK` イベントソースタイプを説明するオブジェクトです。詳細については、「 *AWS Lambda デベロッパーガイド*」の[「Amazon MSK AWS Lambda での の使用](https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html)」を参照してください。
AWS Serverless Application Model (AWS SAM) このイベントタイプが設定されている場合、 は [https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html)リソースを生成します。
Schema Registry を使用するには、関数に特定の IAM ロール許可を定義する必要があります。必要な設定の例については、「[Complete setup with IAM roles](#sam-property-function-msk-example-complete)」を参照してください。
## 構文
AWS SAM テンプレートでこのエンティティを宣言するには、次の構文を使用します。
### YAML
```
[BatchSize](#sam-function-msk-batchsize): Integer
[BisectBatchOnFunctionError](#sam-function-msk-bisectbatchonfunctionerror): Boolean
[ConsumerGroupId](#sam-function-msk-consumergroupid): String
DestinationConfig: [DestinationConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-destinationconfig.html)
[Enabled](#sam-function-msk-enabled): Boolean
[FilterCriteria](#sam-function-msk-filtercriteria): [FilterCriteria](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-filtercriteria.html)
[FunctionResponseTypes](#sam-function-msk-functionresponsetypes): List
KmsKeyArn: String
[MaximumBatchingWindowInSeconds](#sam-function-msk-maximumbatchingwindowinseconds): Integer
[MaximumRecordAgeInSeconds](#sam-function-msk-maximumrecordageinseconds): Integer
[MaximumRetryAttempts](#sam-function-msk-maximumretryattempts): Integer
[LoggingConfig](#sam-function-msk-loggingconfig): [LoggingConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-loggingconfig.html)
[MetricsConfig](#sam-function-msk-metricsconfig): [MetricsConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-metricsconfig.html)
[ProvisionedPollerConfig](#sam-function-msk-provisionedpollerconfig): [ProvisionedPollerConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-provisionedpollerconfig)
[SchemaRegistryConfig](#sam-function-msk-schemaregistryconfig): [SchemaRegistryConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-schemaregistryconfig.html)
SourceAccessConfigurations: [SourceAccessConfigurations](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-sourceaccessconfigurations)
[StartingPosition](#sam-function-msk-startingposition): String
StartingPositionTimestamp: Double
[Stream](#sam-function-msk-stream): String
[Topics](#sam-function-msk-topics): List
```
## プロパティ
`BatchSize`
Lambda がストリームまたはキューから取り出し、関数に送信する各バッチ内の最大レコード数。Lambda は、同期呼び出しのペイロード上限 (6 MB) まで、バッチ内のすべてのレコードを単一の呼び出しで関数に渡します。
*デフォルト*: 100
*有効範囲*: 最小値 1。最大値は 10,000 です。
*タイプ:* 整数
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[BatchSize](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-batchsize)`プロパティに直接渡されます。
`BisectBatchOnFunctionError`
関数がエラーを返す場合は、バッチを 2 つに分割して再試行します。
型: ブール
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[BisectBatchOnFunctionError](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-bisectbatchonfunctionerror)`プロパティに直接渡されます。
`ConsumerGroupId`
Kafka トピックからイベントを読み取る方法を設定する文字列。
*タイプ*: 文字列
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[AmazonManagedKafkaConfiguration](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html)`プロパティに直接渡されます。
`DestinationConfig`
Lambda がイベントを処理した後のイベントの送信先を指定する構成オブジェクト。
このプロパティを使用して、Amazon MSK イベントソースから失敗した呼び出しの送信先を指定します。
*タイプ*: [DestinationConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-destinationconfig)
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[ DestinationConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-destinationconfig.html)`プロパティに直接渡されます。
`Enabled`
ポーリングと呼び出しを一時停止するために、イベントソースマッピングを無効にします。
型: ブール
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[Enabled](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-enabled)`プロパティに直接渡されます。
`FilterCriteria`
Lambda がイベントを処理する必要があるかどうかを判断する基準を定義するオブジェクト。詳細については、*AWS Lambda デベロッパーガイド*の [AWS Lambda イベントのフィルタリング](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html)を参照してください。
*タイプ*: [FilterCriteria](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-filtercriteria.html)
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[FilterCriteria](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-filtercriteria.html)`プロパティに直接渡されます。
`FunctionResponseTypes`
現在イベントソースマッピングに適用されているレスポンスタイプのリストです。詳細については、「AWS Lambda デベロッパーガイド」の「[バッチアイテムの失敗をレポートする](https://docs.aws.amazon.com/lambda/latest/dg/kafka-retry-configurations.html)」を参照してください。
*有効な値:* `ReportBatchItemFailures`
*タイプ*: リスト
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[FunctionResponseTypes](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-functionresponsetypes)`プロパティに直接渡されます。
`KmsKeyArn`
このイベントに関連する情報を暗号化するためのキーの Amazon リソースネーム (ARN)。
*タイプ*: 文字列
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[KmsKeyArn](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-kmskeyarn)`プロパティに直接渡されます。
`MaximumBatchingWindowInSeconds`
関数を呼び出すまでのレコード収集の最大時間 (秒) です。
*タイプ:* 整数
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[MaximumBatchingWindowInSeconds](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-maximumbatchingwindowinseconds)`プロパティに直接渡されます。
`MaximumRecordAgeInSeconds`
Lambda が処理のために関数に送信するレコードの最大存続時間です。
*タイプ:* 整数
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[MaximumRecordAgeInSeconds](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-maximumrecordageinseconds)`プロパティに直接渡されます。
`MaximumRetryAttempts`
関数がエラーを返すときの最大再試行回数です。
*タイプ:* 整数
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[MaximumRetryAttempts](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-maximumretryattempts)`プロパティに直接渡されます。
`LoggingConfig`
イベントソースマッピングのログ記録設定を指定する設定オブジェクト。
*Type*: [LoggingConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-loggingconfig.html)
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[LoggingConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-loggingconfig.html)`プロパティに直接渡されます。
`MetricsConfig`
イベントソースマッピングのメトリクス設定を指定する設定オブジェクト。
*タイプ*: [MetricsConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-metricsconfig.html)
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[MetricsConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-metricsconfig.html)`プロパティに直接渡されます。
`ProvisionedPollerConfig`
イベントソースマッピングの計算に使用されるポーラーの数を増やすための設定。この設定では、最小 1 つのポーラーと最大 2000 のポーラーを使用できます。例については、「[ProvisionedPollerConfig の例](#sam-property-function-msk-example-provisionedpollerconfig)」を参照してください。
*タイプ*: [ProvisionedPollerConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-provisionedpollerconfig)
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[ProvisionedPollerConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-provisionedpollerconfig)`プロパティに直接渡されます。
`SchemaRegistryConfig`
Kafka イベントソースでスキーマレジストリを使用するための設定。
この機能は `ProvisionedPollerConfig` を設定するために必要です。
*タイプ*: SchemaRegistryConfig
*必須:* いいえ
*CloudFormation 互換性:* このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[AmazonManagedKafkaEventSourceConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-amazonmanagedkafkaeventsourceconfig)`プロパティに直接渡されます。
`SourceAccessConfigurations`
認証プロトコルの配列、VPC コンポーネント、イベントソースを保護して定義する仮想ホスト。
*有効な値:* `CLIENT_CERTIFICATE_TLS_AUTH`
型: [SourceAccessConfiguration](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-sourceaccessconfiguration.html) のリスト
*必須:* いいえ
*CloudFormation 互換性:* このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの [AmazonManagedKafkaEventSourceConfig](https://docs.aws.amazon.com/AWSCloudFormation/latest/TemplateReference/aws-properties-lambda-eventsourcemapping-amazonmanagedkafkaeventsourceconfig) プロパティの一部です。
`StartingPosition`
読み取りを開始するストリームの場所です。
+ `AT_TIMESTAMP` - レコードの読み取りを開始する時間を指定します。
+ `LATEST` - 新しいレコードのみを読み込みます。
+ `TRIM_HORIZON` - 使用可能なすべてのレコードを処理します。
有効な値: `AT_TIMESTAMP` \$1 `LATEST` \$1 `TRIM_HORIZON`
*タイプ*: 文字列
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[StartingPosition](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-startingposition)`プロパティに直接渡されます。
`StartingPositionTimestamp`
Unix タイム秒単位で読み取りをスタートする時間。`StartingPosition` が `AT_TIMESTAMP` として指定されている場合の `StartingPositionTimestamp` を定義します。
型: 倍精度
*必須:* いいえ
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[StartingPositionTimestamp](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-startingpositiontimestamp)`プロパティに直接渡されます。
`Stream`
データストリームまたはストリームコンシューマーの Amazon リソースネーム (ARN) です。
*タイプ*: 文字列
*必須*: はい
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[EventSourceArn](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-eventsourcearn)`プロパティに直接渡されます。
`Topics`
Kafka トピックの名前です。
*タイプ*: リスト
*必須:* はい
*CloudFormation 互換性*: このプロパティは、 `AWS::Lambda::EventSourceMapping`リソースの `[Topics](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-topics)`プロパティに直接渡されます。
## 例
### IAM ロールを使用してセットアップを完了する
次の例は、Schema Registry を使用するために必要な IAM ロール設定を含む完全なセットアップを示しています。
```
Parameters:
PreCreatedSubnetOne:
Type: String
PreCreatedSubnetTwo:
Type: String
MskClusterName4:
Type: String
Resources:
MyLambdaExecutionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17 '
Statement:
- Action: [sts:AssumeRole]
Effect: Allow
Principal:
Service: [lambda.amazonaws.com]
Policies:
- PolicyName: KafkaClusterPermissions
PolicyDocument:
Statement:
- Action: [kafka:DescribeClusterV2, kafka:GetBootstrapBrokers]
Effect: Allow
Resource: 'arn:aws:kafka:us-east-1:123456789012:cluster/*'
- PolicyName: KafkaAuthPolicy
PolicyDocument:
Statement:
- Action: [secretsmanager:GetSecretValue, kms:Decrypt]
Effect: "Allow"
Resource: ['arn:aws:secretsmanager:us-west-2:123456789012:secret:kafkaSecret-******',
'arn:aws:kms:us-west-2:123456789012:key/keyId']
- PolicyName: ENIPolicy
PolicyDocument:
Statement:
- Action: [ec2:CreateNetworkInterface,
ec2:DescribeNetworkInterfaces, ec2:DescribeVpcs, ec2:DeleteNetworkInterface,
ec2:DescribeSubnets, ec2:DescribeSecurityGroups]
Effect: Allow
Resource: '*'
- PolicyName: SchemaRegistryPolicy
PolicyDocument:
Statement:
- Action: [glue:GetRegistry]
Effect: Allow
Resource: 'arn:aws:glue:{region}:{account-id}:registry/{registry-name}'
- PolicyName: SchemaVersionsPolicy
PolicyDocument:
Statement:
- Action: [glue:GetSchemaVersions]
Effect: Allow
Resource: '*'
ManagedPolicyArns:
- !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
Tags:
- {Value: SAM, Key: lambda:createdBy}
MyMskCluster:
Type: AWS::MSK::Cluster
Properties:
BrokerNodeGroupInfo:
ClientSubnets:
- Ref: PreCreatedSubnetOne
- Ref: PreCreatedSubnetTwo
InstanceType: kafka.t3.small
StorageInfo:
EBSStorageInfo:
VolumeSize: 1
ClusterName:
Ref: MskClusterName4
KafkaVersion: 3.8.x
NumberOfBrokerNodes: 2
MyMskStreamProcessor:
Type: AWS::Serverless::Function
Properties:
Runtime: nodejs18.x
Handler: index.handler
CodeUri: ${codeuri}
Role:
Fn::GetAtt: [MyLambdaExecutionRole, Arn]
Events:
MyMskEvent:
Type: MSK
Properties:
StartingPosition: LATEST
Stream:
Ref: MyMskCluster
SourceAccessConfigurations:
- Type: SASL_SCRAM_512_AUTH
URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
Topics:
- SchemaRegistryTestTopic
ProvisionedPollerConfig:
MinimumPollers: 1
SchemaRegistryConfig:
AccessConfigs:
- Type: BASIC_AUTH
URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
SchemaValidationConfigs:
- Attribute: KEY
EventRecordFormat: JSON
SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry
```
### ProvisionedPollerConfig の例
```
ProvisionedPollerConfig:
MinimumPollers: 1
MaximumPollers: 200
```
### 既存のクラスターの Amazon MSK の例
以下は、 AWS アカウントに既に存在する Amazon MSK クラスター用の `MSK` イベントソースタイプの例です。
#### YAML
```
Events:
MSKEvent:
Type: MSK
Properties:
StartingPosition: LATEST
Stream: arn:aws:kafka:us-east-1:012345678012:cluster/exampleClusterName/abcdefab-1234-abcd-5678-cdef0123ab01-2
Topics:
- MyTopic
```
### 同じテンプレートで宣言されたクラスターの Amazon MSK の例
以下は、同じテンプレートファイルで宣言されている Amazon MSK クラスター用の `MSK` イベントソースタイプの例です。
#### YAML
```
Events:
MSKEvent:
Type: MSK
Properties:
StartingPosition: LATEST
Stream:
Ref: MyMskCluster # This must be the name of an MSK cluster declared in the same template file
Topics:
- MyTopic
```
#### スキーマレジストリを使用した MSK イベントソース
以下は、スキーマレジストリを使用して設定された `MSK` イベントソースタイプの例です。
```
Events:
MSKEvent:
Type: MSK
Properties:
StartingPosition: LATEST
Stream:
Ref: MyMskCluster
Topics:
- SchemaRegistryTestTopic
ProvisionedPollerConfig:
MinimumPollers: 1
SchemaRegistryConfig:
SchemaRegistryURI: !Sub arn:${AWS::Partition}:glue:us-west-2:123456789012:registry/myregistry
EventRecordFormat: JSON
SchemaValidationConfigs:
- Attribute: KEY
- Attribute: VALUE
```
#### Confluent Schema Registry を使用した MSK イベントソース
以下は、Confluent Schema Registry を使用して設定された `MSK` イベントソースタイプの例です。
```
Events:
MSKEvent:
Type: MSK
Properties:
StartingPosition: LATEST
Stream:
Ref: MyMskCluster
Topics:
- SchemaRegistryTestTopic
ProvisionedPollerConfig:
MinimumPollers: 1
SchemaRegistryConfig:
SchemaRegistryURI: https://my-schema-registry.confluent.cloud
AccessConfigs:
- Type: BASIC_AUTH
URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-secret
EventRecordFormat: JSON
SchemaValidationConfigs:
- Attribute: KEY
- Attribute: VALUE
```