本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# MSK
描述 `MSK` 事件源类型的对象。有关更多信息，请参阅*AWS Lambda 开发者指南*中的[AWS Lambda 与 Amazon MSK 搭配使用](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)资源。
要使用架构注册表，您需要为函数定义特定的 IAM 角色权限。有关所需配置的示例，请参阅[使用 IAM 角色完成设置](#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
```
## Properties
`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`
如果函数返回错误，则将批次拆分为两批并重试。
*类型*：布尔值
*必需*：否
*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`
一个配置对象，用于指定事件源映射的日志配置。
*类型*：[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 角色完成设置
以下示例展示了完整的配置，包括使用架构注册表所需的 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 架构注册表的 MSK 事件源
以下是一个配置了 Confluent 架构注册表的 `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
```