(lambda-event-sources): Support MSK and self hosted Kafka

[bracki]

Add a KafkaEventSource that implements StreamEventSource and allows configuring either MSK or a self hosted Kafka cluster as described here and here.

Use Case

I want to setup MSK/Kafka as event source for Lambda without doing it with the lower level CfnEventSourceMapping.

Proposed Solution

Add a class KafkaEventSource that implements StreamEventSource.

• 👋 I may be able to implement this feature request
• ⚠️ This feature might incur a breaking change

---

This is a 🚀 Feature Request

[bracki]

It could look like this:

export type KafkaEventSourceProps = StreamEventSourceProps

class KafkaEventSource extends StreamEventSource {
    constructor(readonly clusterArn: string,
                readonly kafkaTopic: string,
                readonly secretArn: string,
                readonly kmsKeyArn: string,
                props: KafkaEventSourceProps) {
        super(props);
    }

    public bind(target: lambda.IFunction) {
        const eventSourceMapping = target.addEventSourceMapping(
            `KafkaEventSource:${this.kafkaTopic}`,
            this.enrichMappingOptions({
                eventSourceArn: this.clusterArn,
                kafkaTopic: this.kafkaTopic,
                startingPosition: this.props.startingPosition
            })
        );

        target.addToRolePolicy(new iam.PolicyStatement(
            {
                actions: ['secretsmanager:GetSecretValue'],
                resources: [this.secretArn],
            })
        );
        target.addToRolePolicy(new iam.PolicyStatement(
            {
                actions: ['kms:Decrypt'],
                resources: [this.kmsKeyArn],
            })
        );
        target.addToRolePolicy(new iam.PolicyStatement(
            {
                actions: ['kafka:DescribeCluster', 'kafka:GetBootstrapBrokers', 'kafka:ListScramSecrets'],
                resources: [this.clusterArn]
            }
        ));
        target.role?.addManagedPolicy(iam.ManagedPolicy.fromAwsManagedPolicyName('service-role/AWSLambdaMSKExecutionRole'));

        const cfnEventSourceMapping = eventSourceMapping.node.defaultChild as lambda.CfnEventSourceMapping;
        cfnEventSourceMapping.sourceAccessConfigurations = [{type: 'SASL_SCRAM_512_AUTH', uri: this.secretArn}];
    }
}

[bracki]

For a self hosted Kafka cluster this is needed as well: aws-cloudformation/cloudformation-coverage-roadmap#723 Or it could be retrofitted with AwsCustomResource.

Maybe we have a quick chat @nija-at?

[nija-at]

If it's supported by the AWS SDK for Javascript, you could absolutely use a custom resource until support for it is available via CloudFormation.

@bracki - I'm out on vacation for the next 2 weeks but feel free to take a stab at a PR here. You can DM me (once I'm back) on the public CDK Slack channel (details at https://cdk.dev/)

[github-actions]

⚠️COMMENT VISIBILITY WARNING⚠️

Comments on closed issues are hard for our team to see.
If you need more assistance, please either tag a team member or open a new issue that references this one.
If you wish to keep having a conversation with other community members under this issue feel free to do so.