feat(lambda-event-sources): adds AuthenticationMethod.CLIENT_CERTIFICATE_TLS_AUTH to kafka (#17920)

This PR adds a new enum value, `CLIENT_CERTIFICATE_TLS_AUTH`, to `SelfManagedKafkaEventSource`.

[Docs](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-sourceaccessconfiguration.html).

----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: oieduardorabelo

packages/@aws-cdk/aws-lambda-event-sources/lib/kafka.ts

@@ -53,6 +53,10 @@ export enum AuthenticationMethod {
    * BASIC_AUTH (SASL/PLAIN) authentication method for your Kafka cluster
    */
   BASIC_AUTH = 'BASIC_AUTH',
+  /**
+   * CLIENT_CERTIFICATE_TLS_AUTH (mTLS) authentication method for your Kafka cluster
+   */
+  CLIENT_CERTIFICATE_TLS_AUTH = 'CLIENT_CERTIFICATE_TLS_AUTH',
 }
 
 /**
@@ -213,6 +217,9 @@ export class SelfManagedKafkaEventSource extends StreamEventSource {
       case AuthenticationMethod.BASIC_AUTH:
         authType = lambda.SourceAccessConfigurationType.BASIC_AUTH;
         break;
+      case AuthenticationMethod.CLIENT_CERTIFICATE_TLS_AUTH:
+        authType = lambda.SourceAccessConfigurationType.CLIENT_CERTIFICATE_TLS_AUTH;
+        break;
       case AuthenticationMethod.SASL_SCRAM_256_AUTH:
         authType = lambda.SourceAccessConfigurationType.SASL_SCRAM_256_AUTH;
         break;

packages/@aws-cdk/aws-lambda-event-sources/test/kafka.test.ts

@@ -489,6 +489,41 @@ describe('KafkaEventSource', () => {
       });
     });
 
+    test('using CLIENT_CERTIFICATE_TLS_AUTH', () => {
+      // GIVEN
+      const stack = new cdk.Stack();
+      const fn = new TestFunction(stack, 'Fn');
+      const kafkaTopic = 'some-topic';
+      const secret = new Secret(stack, 'Secret', { secretName: 'AmazonMSK_KafkaSecret' });
+      const bootstrapServers = ['kafka-broker:9092'];
+      const sg = SecurityGroup.fromSecurityGroupId(stack, 'SecurityGroup', 'sg-0123456789');
+      const vpc = new Vpc(stack, 'Vpc');
+
+      // WHEN
+      fn.addEventSource(new sources.SelfManagedKafkaEventSource(
+        {
+          bootstrapServers: bootstrapServers,
+          topic: kafkaTopic,
+          secret: secret,
+          startingPosition: lambda.StartingPosition.TRIM_HORIZON,
+          vpc: vpc,
+          vpcSubnets: { subnetType: SubnetType.PRIVATE_WITH_NAT },
+          securityGroup: sg,
+          authenticationMethod: sources.AuthenticationMethod.CLIENT_CERTIFICATE_TLS_AUTH,
+        }));
+
+      Template.fromStack(stack).hasResourceProperties('AWS::Lambda::EventSourceMapping', {
+        SourceAccessConfigurations: Match.arrayWith([
+          {
+            Type: 'CLIENT_CERTIFICATE_TLS_AUTH',
+            URI: {
+              Ref: 'SecretA720EF05',
+            },
+          },
+        ]),
+      });
+    });
+
     test('ManagedKafkaEventSource name conforms to construct id rules', () => {
       // GIVEN
       const stack = new cdk.Stack();

packages/@aws-cdk/aws-lambda/lib/event-source-mapping.ts

@@ -35,6 +35,12 @@ export class SourceAccessConfigurationType {
    */
   public static readonly SASL_SCRAM_512_AUTH = new SourceAccessConfigurationType('SASL_SCRAM_512_AUTH');
 
+  /**
+   * The Secrets Manager ARN of your secret key containing the certificate chain (X.509 PEM), private key (PKCS#8 PEM),
+   * and private key password (optional) used for mutual TLS authentication of your MSK/Apache Kafka brokers.
+   */
+  public static readonly CLIENT_CERTIFICATE_TLS_AUTH = new SourceAccessConfigurationType('CLIENT_CERTIFICATE_TLS_AUTH');
+
   /** A custom source access configuration property */
   public static of(name: string): SourceAccessConfigurationType {
     return new SourceAccessConfigurationType(name);