教程：使用 Amazon MSK 事件源映射调用 Lambda 函数

为您的接口 Amazon VPC 端点创建一个安全组 endpointSecurityGroup，以允许来自 clusterSecurityGroups 端口 443 的入站 TCP 流量。按照 Amazon EC2 文档中创建安全组的步骤创建安全组。然后，按照 Amazon EC2 文档中向安全组添加规则的步骤添加相应的规则。

使用以下信息创建安全组：

添加入站规则时，为 clusterSecurityGroups 中的每个安全组创建一条规则。对于每条规则：

创建一个端点，将 Lambda 服务连接到包含 Amazon MSK 集群的 Amazon VPC。按照创建接口端点中的步骤进行操作。

使用以下信息创建接口端点：

创建一个端点，将 Amazon STS 服务连接到包含 Amazon MSK 集群的 Amazon VPC。按照创建接口端点中的步骤进行操作。

使用以下信息创建接口端点：

对于与您的 Amazon MSK 集群关联的每个安全组（即 clusterSecurityGroups），允许执行以下操作：

默认安全组规则允许部分流量，因此，如果您的集群连接到单个安全组，并且该组有默认规则，则不需要其他规则。要调整安全组规则，请按照 Amazon EC2 文档中向安全组添加规则的步骤进行操作。

使用以下信息向您的安全组添加规则：

编写一个 IAM 策略文档（JSON 文档）clusterAuthPolicy，允许 Lambda 使用您的 Kafka 使用者组从 Amazon MSK 集群中的 Kafka 主题进行读取。Lambda 要求在读取时设置一个 Kafka 使用者组。

修改以下模板以符合您的先决条件：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:DescribeGroup",
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:ReadData",
                "kafka-cluster:DescribeClusterDynamicConfiguration"
            ],
            "Resource": [
                "arn:aws:kafka:region:account-id:cluster/mskClusterName/cluster-uuid",
                "arn:aws:kafka:region:account-id:topic/mskClusterName/cluster-uuid/mskTopicName",
                "arn:aws:kafka:region:account-id:group/mskClusterName/cluster-uuid/mskGroupName"
            ]
        }
    ]
}       

有关更多信息，请参阅 基于 IAM 角色的身份验证。编写策略时：

确定 Lambda 发现和连接您的 Amazon MSK 集群所需的 Amazon MSK、Amazon EC2 和 CloudWatch 权限，并记录这些事件。

AWSLambdaMSKExecutionRole 托管策略宽松地定义所需的权限。在以下步骤中使用该策略。

在生产环境中，评测 AWSLambdaMSKExecutionRole 以根据最低权限原则限制您的执行角色策略，然后为您的角色编写一个策略来取代此托管策略。

登录 Amazon Web Services Management Console，然后使用以下网址打开 IAM 控制台：https://console.aws.amazon.com/iam/。

在左侧的导航窗格中，选择策略。

选择创建策略。

在策略编辑器部分，选择 JSON 选项。

粘贴 clusterAuthPolicy。

向策略添加完权限后，选择下一步。

在查看和创建页面上，为创建的策略键入策略名称和描述（可选）。查看此策略中定义的权限以查看策略授予的权限。

选择创建策略可保存新策略。

在 IAM 控制台中，打开 Roles（角色）页面。

选择 Create role（创建角色）。

在可信实体类型下，选择 Amazon 服务。

在 Use case（使用案例）下，选择 Lambda。

选择下一步。

选择以下策略：

选择下一步。

对于角色名称，输入 lambdaAuthRole，然后选择创建角色。

打开 Lambda 控制台并从标题中选择创建函数。

选择从头开始编写。

对于函数名称，请提供您选择的相应名称。

对于运行时，选择最新支持版本的 Node.js 以使用本教程中提供的代码。

选择更改默认执行角色。

选择使用现有角色。

对于现有角色，选择 lambdaAuthRole。

导航到您的 Lambda 函数的概述页面。

在函数概述部分中，选择左下角的添加触发器。

在选择源下拉列表中，选择 Amazon MSK。

请勿设置身份验证。

对于 MSK 集群，选择集群的名称。

对于批次大小，输入 1。此步骤使该功能更易于测试，但并非生产中的理想值。

对于主题名称，输入 Kafka 主题名称。

对于使用者组 ID，请提供您的 Kafka 使用者组的 ID。

导航到您的 Lambda 函数的概述页面。

选择节点选项卡。

将提供的代码输入到代码源 IDE 中。

在代码源导航栏中，选择部署。

使用您的 Kafka 管理主机通过 CLI kafka-console-producer 生成 Kafka 事件。有关更多信息，请参阅 Kafka 文档中的 Write some events into the topic。发送足够的事件以填充上一步中定义的事件源映射批次大小所定义的批次，否则 Lambda 将等待更多信息来调用。

如果您的函数运行，则 Lambda 会将发生的事件写入 CloudWatch。在控制台中，导航到您的 Lambda 函数的详细信息页面。

选择 Configuration（配置）选项卡。

在侧栏中，选择监控和操作工具。

在日志记录配置下确定 CloudWatch 日志组。日志组应以 /aws/lambda 开头。选择日志组的链接。

在 CloudWatch 控制台中，检查日志事件，查看 Lambda 已发送到日志流的日志事件。确定是否存在包含来自 Kafka 事件消息的日志事件，如下图所示。如果有，则表示您已成功使用 Lambda 事件源映射将 Lambda 函数连接到 Amazon MSK。