Lamba_python_cloudwatch_msk.py

import base64
import json
import boto3
from kafka import KafkaProducer

msk_topic = 'my-msk-topic'
msk_broker_list = ['my-msk-broker-1:9092', 'my-msk-broker-2:9092']  # Replace with your own broker list

# Create a Kafka producer with the specified broker list
producer = KafkaProducer(bootstrap_servers=msk_broker_list)

def lambda_handler(event, context):
    # Get the raw log data from the CloudWatch subscription
    log_data = base64.b64decode(event['awslogs']['data'])
    logs = json.loads(log_data)
    
    # Extract the log events and publish them to the Kafka topic
    for log_event in logs['logEvents']:
        message = log_event['message']
        producer.send(msk_topic, message.encode('utf-8'))
    
    # Flush any remaining messages in the producer queue
    producer.flush()