Tried receiving messages in Lambda from another account using Amazon MSK's managed VPC connection

			
			AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: "SAM Application with MSK Managed Connection"

Parameters:
  MSKAccountId:
    Description: AWS Account Id where MSK cluster is deployed
    Type:  String
  MSKConnectionArn:
    Description: ARN of the MSK Managed Connection
    Default: ""
    Type: String

Resources:
  # VPC
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.10.0.0/16
      EnableDnsHostnames: true
      EnableDnsSupport: true
      Tags:
        - Key: Name
          Value: uehara-test-vpc

  # Private subnet 1 (AZ: ap-northeast-1a)
  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.10.0.0/26
      AvailabilityZone: ap-northeast-1a
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: uehara-test-private-subnet-1a

  # Private subnet 2 (AZ: ap-northeast-1c)
  PrivateSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.10.0.64/26
      AvailabilityZone: ap-northeast-1c
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: uehara-test-private-subnet-1c

  # Security group for VPC endpoints
  VpcEndpointSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group for VPC endpoints
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 443
          ToPort: 443
          CidrIp: !GetAtt VPC.CidrBlock
      Tags:
        - Key: "Name"
          Value: !Sub "uehara-test-vpc-endpoint-sg"

  # Lambda VPC endpoint (for event source mapping)
  LambdaVPCEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
      VpcId: !Ref VPC
      ServiceName: !Sub 'com.amazonaws.${AWS::Region}.lambda'
      VpcEndpointType: Interface
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
      SecurityGroupIds:
        - !Ref VpcEndpointSecurityGroup
      PrivateDnsEnabled: true

  # STS VPC endpoint (for event source mapping)
  STSVPCEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
      VpcId: !Ref VPC
      ServiceName: !Sub 'com.amazonaws.${AWS::Region}.sts'
      VpcEndpointType: Interface
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
      SecurityGroupIds:
        - !Ref VpcEndpointSecurityGroup
      PrivateDnsEnabled: true

  # Security group for managed connection
  ManagedConnectionSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group for Managed Connection
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 14001
          ToPort: 14100
          CidrIp: !GetAtt VPC.CidrBlock
      SecurityGroupEgress:
        - Description: Allow all outbound traffic
          IpProtocol: "-1"
          CidrIp: 0.0.0.0/0
      Tags:
        - Key: "Name"
          Value: !Sub "uehara-test-managed-connection-sg"

  # Self-referencing configuration
  ManagedConnectionSecurityGroupIngress:
    Type: AWS::EC2::SecurityGroupIngress
    Properties:
      GroupId: !Ref ManagedConnectionSecurityGroup
      IpProtocol: tcp
      FromPort: 14001
      ToPort: 14100
      SourceSecurityGroupId: !Ref ManagedConnectionSecurityGroup

  # Lambda execution role
  MSKProcessorExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: msk-processor-execution-role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: "MSKAccessPolicy"
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - kafka:DescribeVpcConnection
                Resource: '*'
              - Effect: Allow
                Action:
                  - kafka-cluster:Connect
                  - kafka-cluster:DescribeClusterDynamicConfiguration
                Resource: !Sub 'arn:aws:kafka:${AWS::Region}:${MSKAccountId}:cluster/*/*'
              - Effect: Allow
                Action:
                  - kafka-cluster:DescribeTopic
                  - kafka-cluster:ReadData
                Resource: !Sub 'arn:aws:kafka:${AWS::Region}:${MSKAccountId}:topic/*/*'
              - Effect: Allow
                Action:
                  - kafka-cluster:AlterGroup
                  - kafka-cluster:DescribeGroup
                Resource: !Sub 'arn:aws:kafka:${AWS::Region}:${MSKAccountId}:group/*/*'
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
        - arn:aws:iam::aws:policy/service-role/AWSLambdaMSKExecutionRole

  # MSK event source Lambda function
  MSKProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: msk-processor-function
      CodeUri: function/
      Handler: msk_processor.lambda_handler
      Runtime: python3.11
      Architectures:
        - x86_64
      Timeout: 300
      MemorySize: 512
      Role: !GetAtt MSKProcessorExecutionRole.Arn
      Layers:
        - "arn:aws:lambda:ap-northeast-1:336392948345:layer:AWSSDKPandas-Python311:22"
      Environment:
        Variables:
          LOG_LEVEL: INFO

  # # MSK event source mapping
  # MSKEventSourceMapping:
  #   Type: AWS::Lambda::EventSourceMapping
  #   Properties:
  #     EventSourceArn: !Ref MSKConnectionArn
  #     FunctionName: !Ref MSKProcessorFunction
  #     StartingPosition: LATEST
  #     Topics:
  #       - test-topic
  #     BatchSize: 10

Outputs:
  VpcId:
    Description: ID of the created VPC
    Value: !Ref VPC
    Export:
      Name: !Sub "${AWS::StackName}-VpcId"

  PrivateSubnet1Id:
    Description: ID of Private Subnet 1
    Value: !Ref PrivateSubnet1
    Export:
      Name: !Sub "${AWS::StackName}-PrivateSubnet1"

  PrivateSubnet2Id:
    Description: ID of Private Subnet 2
    Value: !Ref PrivateSubnet2
    Export:
      Name: !Sub "${AWS::StackName}-PrivateSubnet2"

  MSKProcessorFunctionArn:
    Description: ARN of the MSK Processor Lambda function
    Value: !GetAtt MSKProcessorFunction.Arn
    Export:
      Name: !Sub "${AWS::StackName}-MSKProcessorFunctionArn"
```:::

Here is one supplementary point: I've intentionally commented out the `MSKEventSourceMapping`.

This is meant to be deployed after setting up the MSK managed VPC connection, which will be explained in the later deployment steps, and is commented out because it shouldn't be deployed during the initial deployment.

I could have put it in a separate file, but for simplicity, I've kept everything in one file.

The created resources are as described in the comments, but as additional information, since the event source mapping operates in a private subnet, I've prepared VPC endpoints for Lambda and STS so that the event source mapping can invoke the Lambda function.

Alternatively, you could also set up a public subnet with a NAT Gateway and configure routing from the private subnet.

Finally, `msk_processor.py` is the script file for the Lambda function that will be triggered by the event source mapping.

```python:msk_processor.py
import base64
import json
import logging
from typing import Any, Dict, List

# Log configuration
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def process_kafka_record(record: Dict[str, Any], topic_name: str) -> Dict[str, Any]:
    try:
        # Decode Kafka message
        if "value" in record:
            # Decode Base64 encoded message
            message_bytes = base64.b64decode(record["value"])
            message_str = message_bytes.decode("utf-8")

            try:
                # Try to parse as JSON
                message_data = json.loads(message_str)
            except json.JSONDecodeError:
                # If not JSON, treat as string
                message_data = message_str
        else:
            message_data = None

        # Extract metadata
        partition = record.get("partition", "unknown")
        offset = record.get("offset", "unknown")
        timestamp = record.get("timestamp", "unknown")

        logger.info(
            f"Processing message from topic '{topic_name}', partition {partition}, offset {offset}"
        )
        logger.info(f"Message content: {message_data}")

        # Implement business logic here

        processed_data = {
            "topic": topic_name,
            "partition": partition,
            "offset": offset,
            "timestamp": timestamp,
            "original_message": message_data,
            "processing_status": "success",
        }

        return processed_data

    except Exception as e:
        logger.error(f"Error processing record: {str(e)}")
        return {
            "topic": topic_name,
            "partition": record.get("partition", "unknown"),
            "offset": record.get("offset", "unknown"),
            "error": str(e),
            "processing_status": "error",
        }

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    logger.info(f"Received event: {json.dumps(event, default=str)}")

    processed_records = []

    try:
        # MSK event structure: records by topic are contained in event['records']
        if "records" in event:
            for topic_name, topic_records in event["records"].items():
                logger.info(f"Processing topic: {topic_name}")

                for record in topic_records:
                    processed_record = process_kafka_record(record, topic_name)
                    processed_records.append(processed_record)

        logger.info(f"Successfully processed {len(processed_records)} records")

        return {
            "statusCode": 200,
            "body": json.dumps(
                {
                    "message": f"Successfully processed {len(processed_records)} records",
                    "processed_records": len(processed_records),
                }
            ),
        }

    except Exception as e:
        logger.error(f"Error processing MSK event: {str(e)}")
        raise
```I am writing with the assumption of receiving events by directly specifying MSK as a Lambda source through event source mapping.

As this is a sample script, no specific business logic is implemented, and it only outputs received messages to the log.

## Deployment Procedure
Once the files are ready, deploy in the following order.

### 1. (Producer side) Configure multi-VPC connection for the MSK cluster

In the Producer account, set "Turn on multi-VPC connectivity" from the network settings for the target MSK cluster.

![0905_msk_02](https://devio2024-media.developers.io/image/upload/v1757065942/2025/09/05/vqpnrrubzxia9pyfpnfa.png)

The authentication type will be IAM role-based authentication.

![0905_msk_03](https://devio2024-media.developers.io/image/upload/v1757066060/2025/09/05/lwnxbalvkgab3w93n7pa.png)

It takes some time for the configuration to complete, so please wait. (In my environment, it took about 40-50 minutes)

Once AWS PrivateLink is activated, the configuration is complete.

![0905_msk_04](https://devio2024-media.developers.io/image/upload/v1757066348/2025/09/05/cnrdvaj2dgahtyynkwor.png)

### 2. (Consumer side) Application deployment (excluding event source mapping)

After AWS PrivateLink is enabled, deploy the Lambda function to the Consumer account.

At this point, do not create the event source mapping yet.

Deploy by setting the Producer account ID parameter as follows:

```bash
$ sam deploy --parameter-overrides "MSKAccountId=<Producer account ID>"

		

			
			# MSK event source mapping
MSKEventSourceMapping:
  Type: AWS::Lambda::EventSourceMapping
  Properties:
    EventSourceArn: !Ref MSKConnectionArn
    FunctionName: !Ref MSKProcessorFunction
    StartingPosition: LATEST
    Topics:
      - test-topic
    BatchSize: 10

		

			
			$ sam deploy --parameter-overrides "MSKAccountId=<Producer account ID>" "MSKConnectionArn=<Managed VPC connection ARN>"

		

			
			echo '{"message": "Hello Kafka"}' | 
./kafka_2.13-3.9.1/bin/kafka-console-producer.sh \
  --bootstrap-server <MSK Bootstrap server hostname>:9098 \
  --producer.config client.properties \
  --topic test-topic