本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。
# 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 Resource Name (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 時間秒為單位。定義 `StartingPositionTimestamp` `StartingPosition` 何時指定為 `AT_TIMESTAMP`。
*類型*：Double
*必要*：否
*CloudFormation 相容性*：此屬性會直接傳遞至 `AWS::Lambda::EventSourceMapping` 資源的 `[StartingPositionTimestamp](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-startingpositiontimestamp)` 屬性。
`Stream`
資料串流或串流取用者的 Amazon Resource Name (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 範例
以下是已存在於 之 Amazon MSK 叢集`MSK`的事件來源類型範例 AWS 帳戶。
#### 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
```