Prerequisites Configure network connectivity for Lambda to communicate with Amazon MSK Create an IAM role for Lambda to read from your Amazon MSK topic Create a Lambda function to read from your Amazon MSK topic Create an event source mapping to your Lambda function Update your Lambda function to read your streaming data Test your Lambda function to verify it is connected to your Amazon MSK topic

Tutorial: Using an Amazon MSK event source mapping to invoke a Lambda function

In this tutorial, you will perform the following:

Create a Lambda function in the same AWS account as an existing Amazon MSK cluster.
Configure networking and authentication for Lambda to communicate with Amazon MSK.
Set up a Lambda Amazon MSK event source mapping, which runs your Lambda function when events show up in the topic.

After you are finished with these steps, when events are sent to Amazon MSK, you will be able to set up a Lambda function to process those events automatically with your own custom Lambda code.

What can you do with this feature?

Example solution: Use an MSK event source mapping to deliver live scores to your customers.

Consider the following scenario: Your company hosts a web application where your customers can view information about live events, such as sports games. Information updates from the game are provided to your team through a Kafka topic on Amazon MSK. You want to design a solution that consumes updates from the MSK topic to provide an updated view of the live event to customers inside an application you develop. You have decided on the following design approach: Your client applications will communicate with a serverless backend hosted in AWS. Clients will connect over websocket sessions using the Amazon API Gateway WebSocket API.

In this solution, you need a component that reads MSK events, performs some custom logic to prepare those events for the application layer and then forwards that information to the API Gateway API. You can implement this component with AWS Lambda, by providing your custom logic in a Lambda function, then calling it with a AWS Lambda Amazon MSK event source mapping.

For more information about implementing solutions using the Amazon API Gateway WebSocket API, see WebSocket API tutorials in the API Gateway documentation.

Prerequisites

An AWS account with the following preconfigured resources:

To fulfill these prerequisites, we recommend following Getting started using Amazon MSK in the Amazon MSK documentation.

An Amazon MSK cluster. See Create an Amazon MSK cluster in Getting started using Amazon MSK.
The following configuration:
- Ensure IAM role-based authentication is Enabled in your cluster security settings. This improves your security by limiting your Lambda function to only access the Amazon MSK resources needed. This is enabled by default on new Amazon MSK clusters.
- Ensure Public access is off in your cluster networking settings. Restricting your Amazon MSK cluster's access to the internet improves your security by limiting how many intermediaries handle your data. This is enabled by default on new Amazon MSK clusters.
A Kafka topic in your Amazon MSK cluster to use for this solution. See Create a topic in Getting started using Amazon MSK.
A Kafka admin host set up to retrieve information from your Kafka cluster and send Kafka events to your topic for testing, such as an Amazon EC2 instance with the Kafka admin CLI and Amazon MSK IAM library installed. See Create a client machine in Getting started using Amazon MSK.

Once you have set up these resources, gather the following information from your AWS account to confirm that you are ready to continue.

The name of your Amazon MSK cluster. You can find this information in the Amazon MSK console.
The cluster UUID, part of the ARN for your Amazon MSK cluster, which you can find in the Amazon MSK console. Follow the procedures in Listing clusters in the Amazon MSK documentation to find this information.
The security groups associated with your Amazon MSK cluster. You can find this information in the Amazon MSK console. In the following steps, refer to these as your clusterSecurityGroups.
The id of the Amazon VPC containing your Amazon MSK cluster. You can find this information by identifying subnets associated with your Amazon MSK cluster in the Amazon MSK console, then identifying the Amazon VPC associated with the subnet in the Amazon VPC Console.
The name of the Kafka topic used in your solution. You can find this information by calling your Amazon MSK cluster with the Kafka topics CLI from your Kafka admin host. For more information about the topics CLI, see Adding and removing topics in the Kafka documentation.
The name of a consumer group for your Kafka topic, suitable for use by your Lambda function. This group can be created automatically by Lambda, so you don't need to create it with the Kafka CLI. If you do need to manage your consumer groups, to learn more about the consumer-groups CLI, see Managing Consumer Groups in the Kafka documentation.

The following permissions in your AWS account:

Permission to create and manage a Lambda function.
Permission to create IAM policies and associate them with your Lambda function.
Permission to create Amazon VPC endpoints and alter networking configuration in the Amazon VPC hosting your Amazon MSK cluster.

Configure network connectivity for Lambda to communicate with Amazon MSK

Use AWS PrivateLink to connect Lambda and Amazon MSK. You can do so by creating interface Amazon VPC endpoints in the Amazon VPC console. For more information about networking configuration, see Configure network security.

When a Amazon MSK event source mapping runs on the behalf of a Lambda function, it assumes the Lambda function’s execution role. This IAM role authorizes the mapping to access resources secured by IAM, such as your Amazon MSK cluster. Although the components share an execution role, the Amazon MSK mapping and your Lambda function have separate connectivity requirements for their respective tasks, as shown in the following diagram.

A Lambda function polls a cluster and communicates with Lambda using AWS STS.

Your event source mapping belongs to your Amazon MSK cluster security group. In this networking step, create Amazon VPC endpoints from your Amazon MSK cluster VPC to connect the event source mapping to the Lambda and STS services. Secure these endpoints to accept traffic from your Amazon MSK cluster security group. Then, adjust the Amazon MSK cluster security groups to allow the event source mapping to communicate with the Amazon MSK cluster.

You can configure the following steps using the AWS Management Console.

To configure interface Amazon VPC endpoints to connect Lambda and Amazon MSK

Create a security group for your interface Amazon VPC endpoints, endpointSecurityGroup, that allows inbound TCP traffic on 443 from clusterSecurityGroups. Follow the procedure in Create a security group in the Amazon EC2 documentation to create a security group. Then, follow the procedure in Add rules to a security group in the Amazon EC2 documentation to add appropriate rules.

Create a security group with the following information:

When adding your inbound rules, create a rule for each security group in clusterSecurityGroups. For each rule:
- For Type, select HTTPS.
- For Source, select one of clusterSecurityGroups.
Create an endpoint connecting the Lambda service to the Amazon VPC containing your Amazon MSK cluster. Follow the procedure in Create an interface endpoint.

Create an interface endpoint with the following information:
- For Service name, select com.amazonaws.regionName.lambda, where regionName hosts your Lambda function.
- For VPC, select the Amazon VPC containing your Amazon MSK cluster.
- For Security groups, select endpointSecurityGroup, which you created earlier.
- For Subnets, select the subnets that host your Amazon MSK cluster.
- For Policy, provide the following policy document, which secures the endpoint for use by the Lambda service principal for the lambda:InvokeFunction action.
```
{
    "Statement": [
        {
            "Action": "lambda:InvokeFunction",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            },
            "Resource": "*"
        }
    ]
}
```
- Ensure Enable DNS name remains set.
Create an endpoint connecting the AWS STS service to the Amazon VPC containing your Amazon MSK cluster. Follow the procedure in Create an interface endpoint.

Create an interface endpoint with the following information:
- For Service name, select AWS STS.
- For VPC, select the Amazon VPC containing your Amazon MSK cluster.
- For Security groups, select endpointSecurityGroup.
- For Subnets, select the subnets that host your Amazon MSK cluster.
- For Policy, provide the following policy document, which secures the endpoint for use by the Lambda service principal for the sts:AssumeRole action.
```
{
    "Statement": [
        {
            "Action": "sts:AssumeRole",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            },
            "Resource": "*"
        }
    ]
}
```
- Ensure Enable DNS name remains set.
For each security group associated with your Amazon MSK cluster, that is, in clusterSecurityGroups, allow the following:
- Allow all inbound and outbound TCP traffic on 9098 to all of clusterSecurityGroups, including within itself.
- Allow all outbound TCP traffic on 443.
Some of this traffic is allowed by default security group rules, so if your cluster is attached to a single security group, and that group has default rules, additional rules are not necessary. To adjust security group rules, follow the procedures in Add rules to a security group in the Amazon EC2 documentation.

Add rules to your security groups with the following information:
- For each inbound rule or outbound rule for port 9098, provide
  - For Type, select Custom TCP.
  - For Port range, provide 9098.
  - For Source, provide one of clusterSecurityGroups.
- For each inbound rule for port 443, for Type, select HTTPS.

Create an IAM role for Lambda to read from your Amazon MSK topic

Identify the auth requirements for Lambda to read from your Amazon MSK topic, then define them in a policy. Create a role, lambdaAuthRole, that authorizes Lambda to use those permissions. Authorize actions on your Amazon MSK cluster using kafka-cluster IAM actions. Then, authorize Lambda to perform Amazon MSK kafka and Amazon EC2 actions needed to discover and connect to your Amazon MSK cluster, as well as CloudWatch actions so Lambda can log what it has done.

To describe the auth requirements for Lambda to read from Amazon MSK

Write an IAM policy document (a JSON document), clusterAuthPolicy, that allows Lambda to read from your Kafka topic in your Amazon MSK cluster using your Kafka consumer group. Lambda requires a Kafka consumer group to be set when reading.

Alter the following template to align with your prerequisites:
```
{
    "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"
            ]
        }
    ]
}       
```
For more information, consult IAM role-based authentication. When writing your policy:
- For region and account-id, provide those that host your Amazon MSK cluster.
- For mskClusterName, provide the name of your Amazon MSK cluster.
- For cluster-uuid, provide the UUID in the ARN for your Amazon MSK cluster.
- For mskTopicName, provide the name of your Kafka topic.
- For mskGroupName, provide the name of your Kafka consumer group.
Identify the Amazon MSK, Amazon EC2 and CloudWatch permissions required for Lambda to discover and connect your Amazon MSK cluster, and log those events.

The AWSLambdaMSKExecutionRole managed policy permissively defines the required permissions. Use it in the following steps.

In a production environment, assess AWSLambdaMSKExecutionRole to restrict your execution role policy based on the principle of least privilege, then write a policy for your role that replaces this managed policy.

For details about the IAM policy language, see the IAM documentation.

Now that you have written your policy document, create an IAM policy so you can attach it to your role. You can do this using the console with the following procedure.

To create an IAM policy from your policy document

Sign in to the AWS Management Console and open the IAM console at https://console.aws.amazon.com/iam/.
In the navigation pane on the left, choose Policies.
Choose Create policy.
In the Policy editor section, choose the JSON option.
Paste clusterAuthPolicy.
When you are finished adding permissions to the policy, choose Next.
On the Review and create page, type a Policy Name and a Description (optional) for the policy that you are creating. Review Permissions defined in this policy to see the permissions that are granted by your policy.
Choose Create policy to save your new policy.

For more information, see Creating IAM policies in the IAM documentation.

Now that you have appropriate IAM policies, create a role and attach them to it. You can do this using the console with the following procedure.

To create an execution role in the IAM console

Open the Roles page in the IAM console.
Choose Create role.
Under Trusted entity type, choose AWS service.
Under Use case, choose Lambda.
Choose Next.
Select the following policies:
- clusterAuthPolicy
- AWSLambdaMSKExecutionRole
Choose Next.
For Role name, enter lambdaAuthRole and then choose Create role.

For more information, see Defining Lambda function permissions with an execution role.

Create a Lambda function to read from your Amazon MSK topic

Create a Lambda function configured to use your IAM role. You can create your Lambda function using the console.

To create a Lambda function using your auth configuration

Open the Lambda console and select Create function from the header.
Select Author from scratch.
For Function name, provide an appropriate name of your choice.
For Runtime, choose the Latest supported version of Node.js to use the code provided in this tutorial.
Choose Change default execution role.
Select Use an existing role.
For Existing role, select lambdaAuthRole.

In a production environment, you usually need to add further policies to the execution role for your Lambda function to meaningfully process your Amazon MSK events. For more information on adding policies to your role, see Add or remove identity permissions in the IAM documentation.

Create an event source mapping to your Lambda function

Your Amazon MSK event source mapping provides the Lambda service the information necessary to invoke your Lambda when appropriate Amazon MSK events occur. You can create a Amazon MSK mapping using the console. Create a Lambda trigger, then the event source mapping is automatically set up.

To create a Lambda trigger (and event source mapping)

Navigate to your Lambda function's overview page.
In the function overview section, choose Add trigger on the bottom left.
In the Select a source dropdown, select Amazon MSK.
Don't set authentication.
For MSK cluster, select your cluster's name.
For Batch size, enter 1. This step makes this feature easier to test, and is not an ideal value in production.
For Topic name, provide the name of your Kafka topic.
For Consumer group ID, provide the id of your Kafka consumer group.

Update your Lambda function to read your streaming data

Lambda provides information about Kafka events through the event method parameter. For an example structure of a Amazon MSK event, see Example event. After you understand how to interpret Lambda forwarded Amazon MSK events, you can alter your Lambda function code to use the information they provide.

Provide the following code to your Lambda function to log the contents of a Lambda Amazon MSK event for testing purposes:

.NET

AWS SDK for .NET