为 Lambda 配置 Amazon MSK 事件源

要使用 Amazon MSK 集群作为 Lambda 函数的事件源，您需要创建连接这两个资源的事件源映射。此页面介绍如何为 Amazon MSK 创建事件源映射。

此页面假设您已经正确配置 MSK 集群及其所在的 Amazon Virtual Private Cloud（VPC）。如果您需要设置集群或 VPC，请参阅为 Lambda 配置 Amazon MSK 集群和 Amazon VPC 网络。

将 Amazon MSK 集群用作事件源

当您添加 Apache Kafka 或 Amazon MSK 集群作为 Lambda 函数的触发器时，该集群将用作事件源。

Lambda 根据您指定的起始位置，从您在 CreateEventSourceMapping 请求中指定为 Topics 的 Kafka 主题读取事件数据。成功进行处理后，会将 Kafka 主题提交给 Kafka 集群。

Lambda 按顺序读取每个 Kafka 主题分区的消息。单个 Lambda 负载可以包含来自多个分区的消息。当有更多记录可用时，Lambda 根据您在 CreateEventSourceMapping 请求中指定的 BatchSize 值，继续对记录进行批处理，直到函数赶上主题的速度。

Lambda 处理各个批次后，会提交该批次中消息的偏移量。如果函数为批处理中的任何消息返回错误，Lambda 将重试整批消息，直到处理成功或消息过期为止。您可以将所有重试都失败的记录发送到失败时的目标，以供日后处理。

为 Amazon MSK 事件源创建事件源映射

要创建事件源映射，您可以使用 Lambda 控制台、Amazon Command Line Interface（CLI）或 Amazon SDK。

以下控制台步骤添加 Amazon MSK 集群作为 Lambda 函数的触发器。这将在后台创建一个事件源映射资源。

您还可以使用 Amazon CLI 中的 create-event-source-mapping 命令创建事件源映射。以下示例创建事件源映射，将 Lambda 函数 my-msk-function 映射到 AWSKafkaTopic 主题，从 LATEST 消息开始。此命令还使用 SourceAccessConfiguration 对象指示 Lambda 在连接到集群时使用 SASL/SCRAM 身份验证。

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 身份验证，请包含指定 CLIENT_CERTIFICATE_TLS_AUTH 的 SourceAccessConfiguration 对象以及 Secrets Manager 密钥 ARN。如下面的命令所示：

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 身份验证时，您不需要 SourceAccessConfiguration 对象。如下面的命令所示：

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

配置集群身份验证方法

Lambda 需要访问 Amazon MSK 集群、检索记录和执行其他任务的权限。Amazon MSK 支持多种方法来使用 MSK 集群进行身份验证。

未经身份验证的访问

如果没有客户端会通过互联网访问集群，则可以使用未经身份验证访问。

SASL/SCRAM 身份验证

Lambda 支持使用 SHA-512 哈希函数和传输层安全性协议（TLS）加密进行简单身份验证和安全层/加盐质疑应答身份验证机制（SASL/SCRAM）身份验证。为使 Lambda 连接到集群，请将身份验证凭证（用户名和密码）存储在 Secrets Manager 密钥中，并在配置事件源映射时引用此密钥。

有关使用 Secrets Manager 的更多信息，请参阅《Amazon Managed Streaming for Apache Kafka Developer Guide》中的 Sign-in credentials authentication with Secrets Manager。

双向 TLS 身份验证

双向 TLS（mTLS）在客户端和服务器之间提供双向身份验证。客户端将证书发送给服务器，以便服务器验证客户端。服务器还将服务器证书发送给客户端，以便客户端验证服务器。

对于 Amazon MSK 与 Lambda 的集成，MSK 集群充当服务器，而 Lambda 充当客户端。

Amazon MSK 不支持自签名服务器证书。Amazon MSK 中的所有代理都使用由 Amazon Trust Services CA 签名的公有证书，默认情况下 Lambda 信任这些证书。

-----BEGIN CERTIFICATE-----  
        <certificate contents>
-----END CERTIFICATE-----      

-----BEGIN PRIVATE KEY-----  
         <private key contents>
-----END PRIVATE KEY-----            

-----BEGIN ENCRYPTED PRIVATE KEY-----  
          <private key contents>
-----END ENCRYPTED PRIVATE KEY-----           

{
 "privateKeyPassword": "testpassword",
 "certificate": "-----BEGIN CERTIFICATE-----
MIIE5DCCAsygAwIBAgIRAPJdwaFaNRrytHBto0j5BA0wDQYJKoZIhvcNAQELBQAw
...
j0Lh4/+1HfgyE2KlmII36dg4IMzNjAFEBZiCRoPimO40s1cRqtFHXoal0QQbIlxk
cmUuiAii9R0=
-----END CERTIFICATE-----
-----BEGIN CERTIFICATE-----
MIIFgjCCA2qgAwIBAgIQdjNZd6uFf9hbNC5RdfmHrzANBgkqhkiG9w0BAQsFADBb
...
rQoiowbbk5wXCheYSANQIfTZ6weQTgiCHCCbuuMKNVS95FkXm0vqVD/YpXKwA/no
c8PH3PSoAaRwMMgOSA2ALJvbRz8mpg==
-----END CERTIFICATE-----",
 "privateKey": "-----BEGIN ENCRYPTED PRIVATE KEY-----
MIIFKzBVBgkqhkiG9w0BBQ0wSDAnBgkqhkiG9w0BBQwwGgQUiAFcK5hT/X7Kjmgp
...
QrSekqF+kWzmB6nAfSzgO9IaoAaytLvNgGTckWeUkWn/V0Ck+LdGUXzAC4RxZnoQ
zp2mwJn2NYB7AZ7+imp0azDZb+8YG2aUCiyqb6PnnA==
-----END ENCRYPTED PRIVATE KEY-----"
} 

有关适用于 Amazon MSK 的 mTLS 的更多信息，请参阅《Amazon Managed Streaming for Apache Kafka Developer Guide》中的 Mutual TLS client authentication for Amazon MSK。

IAM 身份验证

您可以使用 Amazon Identity and Access Management（IAM）来验证连接到 MSK 集群的客户端的身份。使用 IAM 身份验证时，Lambda 依靠函数执行角色中的权限连接到集群、检索记录以及执行其他所需操作。有关包含必要权限的策略示例，请参阅《Amazon Managed Streaming for Apache Kafka Developer Guide》中的 Create authorization policies for the IAM role。

如果 IAM 身份验证在您的 MSK 集群上处于活动状态，并且您没有提供密钥，则 Lambda 自动默认使用 IAM 身份验证。

有关 Amazon MSK 中的 IAM 身份验证的更多信息，请参阅 IAM access control。

Lambda 如何选择引导代理

Lambda 根据集群上可用的身份验证方法以及您是否提供用于身份验证的密钥来选择引导代理。如果您为 mTLS 或 SASL/SCRAM 提供了密钥，Lambda 将自动选择该身份验证方法。如果不提供密钥，Lambda 会选择在集群上处于活动状态的最强身份验证方法。下面是 Lambda 选择代理的优先级顺序，从最强到最弱的身份验证：

可自定义的使用者组 ID

将 Kafka 设置为事件源时，您可以指定使用者组 ID。此使用者组 ID 是您希望 Lambda 函数加入的 Kafka 使用者组的现有标识符。您可以使用此功能将任何正在进行的 Kafka 记录处理设置从其他使用者无缝迁移到 Lambda。

Kafka 向使用者组中的所有使用者分发消息。如果您指定的使用者组 ID 具有其他活动使用者，则 Lambda 将仅接收来自 Kafka 主题的部分消息。如果希望 Lambda 处理主题中的所有消息，请关闭该使用者组中的任何其他使用者。

此外，如果指定了使用者组 ID，而 Kafka 找到了具有相同 ID 的有效现有使用者组，则 Lambda 会忽略事件源映射的 StartingPosition 参数。相反，Lambda 开始根据使用者组的已提交偏移量处理记录。如果指定了使用者组 ID，而 Kafka 找不到现有使用者组，则 Lambda 会使用指定的 StartingPosition 配置事件源。

在所有 Kafka 事件源中，您指定的使用者组 ID 必须是唯一的。在使用指定的使用者组 ID 创建 Kafka 事件源映射后，无法更新此值。

轮询和流的起始位置

StartingPosition 参数告诉 Lambda 何时开始从流中读取消息。有三个选项可供选择：

事件源映射创建或更新期间的流轮询最终是一致的：

此行为意味着，如果您指定 LATEST 作为流的起始位置，事件源映射可能会在创建或更新期间错过事件。为确保不会错过任何事件，请指定 TRIM_HORIZON 或 AT_TIMESTAMP。

事件轮询器扩展模式

您可以为 Kafka 事件源映射选择两种事件轮询器扩展模式之一：

按需模式（默认）

当您最初创建 Amazon MSK 事件源时，Lambda 会分配默认数量的事件轮询器来处理 Kafka 主题中的所有分区。Lambda 根据消息负载自动扩展或缩减事件轮询器的数量。

Lambda 会按一分钟的间隔时间来评估主题中所有分区的偏移滞后。如果偏移延迟太高，则分区接收消息的速度比 Lambda 处理消息的速度更快。如有必要，Lambda 会在主题中添加或删除事件轮询器。添加或删除事件轮询器的自动扩缩过程在评估后的三分钟内发生。

如果目标 Lambda 函数受到限制，Lambda 会减少事件轮询器的数量。此操作通过减少事件轮询器可以检索和发送到函数的消息数来减少函数的工作负载。

预调配模式

对于需要微调事件源映射吞吐量的工作负载，您可以使用预调配模式。在预调配模式下，您可以为预调配事件轮询器数量定义最小和最大限制。这些预调配事件轮询器专用于事件源映射，并且可以通过响应式自动扩缩处理意外的消息激增。对于具有严格性能要求的 Kafka 工作负载，我们建议您使用预调配模式。

在 Lambda 中，事件轮询器是一种能够处理高达 5 Mbps 吞吐量的计算单位。作为参考，假设您的事件源产生的平均有效载荷为 1 MB，并且平均函数持续时间为 1 秒。如果有效载荷未进行任何转换（例如筛选），则单个轮询器可以支持 5 Mbps 的吞吐量和 5 次并发 Lambda 调用。使用预调配模式会产生额外成本。有关定价估算，请参阅 Amazon Lambda 定价。

在预调配模式下，最小事件轮询器数量 (MinimumPollers) 的可接受值范围介于 1 到 200 之间（含首尾）。事件轮询器的最大数量 (MaximumPollers) 的可接受值范围介于 1 到 2000 之间（含首尾）。MaximumPollers 必须大于或等于 MinimumPollers。此外，为了保持分区内的有序处理，Lambda 會将 MaximumPollers 限制为主题中的分区数量。

有关选择适当的最小和最大事件轮询器值的更多详细信息，请参阅使用预调配模式时的最佳实践和注意事项。

您可以使用控制台或 Lambda API 为 Amazon MSK 事件源映射配置预调配模式。

您可以使用 EventSourceMappingConfiguration 中的 ProvisionedPollerConfig 对象，以编程方式配置预调配模式。例如，以下 UpdateEventSourceMapping CLI 命令将 MinimumPollers 值配置为 5，将 MaximumPollers 值配置为 100。

aws lambda update-event-source-mapping \
    --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
    --provisioned-poller-config '{"MinimumPollers": 5, "MaximumPollers": 100}'

配置预调配模式后，您可以通过监控 ProvisionedPollers 指标来观测事件轮询器对您的工作负载的使用情况。有关更多信息，请参阅 事件源映射指标。

要禁用预调配模式并返回默认（按需）模式，您可以使用以下 UpdateEventSourceMapping CLI 命令：

aws lambda update-event-source-mapping \
    --uuid a1b2c3d4-5678-90ab-cdef-EXAMPLE11111 \
    --provisioned-poller-config '{}'

使用预调配模式时的最佳实践和注意事项

事件源映射的最小和最大事件轮询器的最佳配置取决于应用程序的性能需求。建议您从默认最小事件轮询器开始，以设定性能配置文件的基准。根据观测到的消息处理模式和所需的性能配置文件调整配置。

对于流量激增且性能需求严格的工作负载，请增加最少的事件轮询器数以处理消息突然激增。要确定所需的最少事件轮询器数，请考虑工作负载的每秒消息数和平均有效载荷大小，并使用单个事件轮询器的吞吐能力（最高 5 Mbps）作为参考。

为了保持分区内的有序处理，Lambda 会将最大事件轮询器数限制为主题中的分区数量。此外，您的事件源映射可以扩展到的最大事件轮询器数取决于函数的并发设置。

激活预调配模式时，更新您的网络设置以删除 Amazon PrivateLink VPC 端点和关联的权限。

创建跨账户事件源映射

您可以使用多 VPC 私有连接将 Lambda 函数连接到不同 Amazon Web Services 账户 中的预置 MSK 集群。多 VPC 连接使用 Amazon PrivateLink，可将所有流量保持在 Amazon 网络内。

要创建跨账户事件源映射，必须先为 MSK 集群配置多 VPC 连接。创建事件源映射时，请使用托管 VPC 连接 ARN 而非集群 ARN，如以下示例所示。CreateEventSourceMapping 操作也因 MSK 集群使用的身份验证类型而异。

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --topics AWSKafkaTopic \
  --starting-position LATEST \
  --function-name my-kafka-function

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --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:444455556666:secret:my-secret"}]'

aws lambda create-event-source-mapping \
  --event-source-arn arn:aws:kafka:us-east-1:111122223333:vpc-connection/444455556666/my-cluster-name/51jn98b4-0a61-46cc-b0a6-61g9a3d797d5-7 \
  --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:444455556666:secret:my-secret"}]'

所有 Amazon MSK 事件源配置参数

所有 Lambda 事件源类型共享相同的 CreateEventSourceMapping 和 UpdateEventSourceMapping API 操作。但是，只有部分参数适用于 Amazon MSK，如下表中所示。