feat(lambda): support for schema registry for kafka (#34746)

### Reason for this change

Lambda is introducing a new property in Event Sources named `SchemaRegistryConfig` in `SelfManagedKafkaEventSourceConfig` and `AmazonManagedKafkaEventSourceConfig` to set configuration settings for a schema registry that will be used to de-serialize the event read from these Kafka event sources. When specified, it allows de-serialization events before they are passed to target function and validation of their format. The users may use a Confluent registry, a self managed registry or AWS Glue Registry. Note, the even source mapping must have `ProvisionedPollerConfig` set (be on provisioned mode) for this feature to be used.

This feature is currently supported for MSK and Self-managed Kafka event sources.

### Description of changes
This new property can be opted in by setting `SchemaRegistryConfig` in `SelfManagedKafkaEventSourceConfig` or `AmazonManagedKafkaEventSourceConfig`. An example is shown bellow:

```
myFunction.addEventSource(new ManagedKafkaEventSource({
  clusterArn,
  topic,
  startingPosition: lambda.StartingPosition.TRIM_HORIZON,
  provisionedPollerConfig: {
    minimumPollers: 1,
    maximumPollers: 3,
  },
  schemaRegistryConfig: {
    schemaRegistryUri: 'https://example.com',
    eventRecordFormat: lambda.EventRecordFormat.JSON,
    accessConfigs: [
      {
        type: lambda.SchemaRegistryAccessConfigType.BASIC_AUTH,
        uri: 'https://example.com',
      },
      ],
    schemaValidationConfigs: [{ attribute: lambda.SchemaValidationAttribute.KEY }],
  },
}));
```

### Describe any new or updated permissions being added

Following IAM permissions will be added to the target function execution role **only if user passed a Glue registry**.
```
{
  Action: 'glue:GetRegistry',
  Effect: 'Allow',
  Resource: {
    'Fn::GetAtt': ['Registry', 'Arn'], // Glue registry ARN
  },
},
{
  Action: [
    'glue:GetSchemaVersion',
    'glue:GetSchema',
  ],
  Effect: 'Allow',
  Resource: [
    {
      'Fn::GetAtt': ['Registry', 'Arn'],
    },
    'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:schema/lambda-gp-test-glue-schema-registry/*',
  ],
},
```


### Description of how you validated changes



Unit tests for each case have been added in the PR. Note, MSK and SMK validations follow the same path so for validations there are only unit tests for MSK cases which should apply for both.

Integration test for both Glue and confluent case have been added for SMK. Since, MSK requires a Kafka cluster in VPC that we typically do not add integration tests for it.

### Checklist
- [x] My code adheres to the [CONTRIBUTING GUIDE](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md) and [DESIGN GUIDELINES](https://github.com/aws/aws-cdk/blob/main/docs/DESIGN_GUIDELINES.md)

----

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

Author: DamFlancoz

packages/@aws-cdk-testing/framework-integ/test/aws-lambda-event-sources/test/integ.kafka-schema-registry.js.snapshot/SchemaRegistryIntegDefaultTestDeployAssertC190A9FD.assets.json

@@ -0,0 +1,196 @@
+{
+ "Resources": {
+  "ConfluentFunctionServiceRole116DBB1D": {
+   "Type": "AWS::IAM::Role",
+   "Properties": {
+    "AssumeRolePolicyDocument": {
+     "Statement": [
+      {
+       "Action": "sts:AssumeRole",
+       "Effect": "Allow",
+       "Principal": {
+        "Service": "lambda.amazonaws.com"
+       }
+      }
+     ],
+     "Version": "2012-10-17"
+    },
+    "ManagedPolicyArns": [
+     {
+      "Fn::Join": [
+       "",
+       [
+        "arn:",
+        {
+         "Ref": "AWS::Partition"
+        },
+        ":iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+       ]
+      ]
+     }
+    ]
+   }
+  },
+  "ConfluentFunctionServiceRoleDefaultPolicy7BF57A6E": {
+   "Type": "AWS::IAM::Policy",
+   "Properties": {
+    "PolicyDocument": {
+     "Statement": [
+      {
+       "Action": [
+        "secretsmanager:DescribeSecret",
+        "secretsmanager:GetSecretValue"
+       ],
+       "Effect": "Allow",
+       "Resource": [
+        {
+         "Ref": "ConfluentClientCertSecret23688D94"
+        },
+        {
+         "Ref": "ConfluentRootCASecret99CAE53B"
+        }
+       ]
+      }
+     ],
+     "Version": "2012-10-17"
+    },
+    "PolicyName": "ConfluentFunctionServiceRoleDefaultPolicy7BF57A6E",
+    "Roles": [
+     {
+      "Ref": "ConfluentFunctionServiceRole116DBB1D"
+     }
+    ]
+   }
+  },
+  "ConfluentFunction5D29A9A6": {
+   "Type": "AWS::Lambda::Function",
+   "Properties": {
+    "Code": {
+     "ZipFile": "exports.handler = async function handler(event) {\n    console.log('event:', JSON.stringify(event, undefined, 2));\n    return { event };\n}"
+    },
+    "Handler": "index.handler",
+    "Role": {
+     "Fn::GetAtt": [
+      "ConfluentFunctionServiceRole116DBB1D",
+      "Arn"
+     ]
+    },
+    "Runtime": "nodejs18.x"
+   },
+   "DependsOn": [
+    "ConfluentFunctionServiceRoleDefaultPolicy7BF57A6E",
+    "ConfluentFunctionServiceRole116DBB1D"
+   ]
+  },
+  "ConfluentFunctionKafkaEventSource46eb851822d1a9f28d63dc4ec210952etesttopicsmkconfluent90EB60AA": {
+   "Type": "AWS::Lambda::EventSourceMapping",
+   "Properties": {
+    "BatchSize": 100,
+    "FunctionName": {
+     "Ref": "ConfluentFunction5D29A9A6"
+    },
+    "ProvisionedPollerConfig": {
+     "MaximumPollers": 3,
+     "MinimumPollers": 1
+    },
+    "SelfManagedEventSource": {
+     "Endpoints": {
+      "KafkaBootstrapServers": [
+       "kafka-broker-1:9092",
+       "kafka-broker-2:9092",
+       "kafka-broker-3:9092"
+      ]
+     }
+    },
+    "SelfManagedKafkaEventSourceConfig": {
+     "ConsumerGroupId": "test-consumer-group-smk-confluent",
+     "SchemaRegistryConfig": {
+      "AccessConfigs": [
+       {
+        "Type": "BASIC_AUTH",
+        "URI": {
+         "Ref": "ConfluentClientCertSecret23688D94"
+        }
+       }
+      ],
+      "EventRecordFormat": "JSON",
+      "SchemaRegistryURI": "https://schema-registry.example.com",
+      "SchemaValidationConfigs": [
+       {
+        "Attribute": "KEY"
+       }
+      ]
+     }
+    },
+    "SourceAccessConfigurations": [
+     {
+      "Type": "CLIENT_CERTIFICATE_TLS_AUTH",
+      "URI": {
+       "Ref": "ConfluentClientCertSecret23688D94"
+      }
+     },
+     {
+      "Type": "SERVER_ROOT_CA_CERTIFICATE",
+      "URI": {
+       "Ref": "ConfluentRootCASecret99CAE53B"
+      }
+     }
+    ],
+    "StartingPosition": "TRIM_HORIZON",
+    "Topics": [
+     "test-topic-smk-confluent"
+    ]
+   }
+  },
+  "ConfluentRootCASecret99CAE53B": {
+   "Type": "AWS::SecretsManager::Secret",
+   "Properties": {
+    "SecretString": "{\"certificate\":\"-----BEGIN CERTIFICATE-----\\n    MIIE5DCCAsygAwIBAgIRAPJdwaFaNRrytHBto0j5BA0wDQYJKoZIhvcNAQELBQAw\\n    cmUuiAii9R0=\\n    -----END CERTIFICATE-----\\n    -----BEGIN CERTIFICATE-----\\n    MIIFgjCCA2qgAwIBAgIQdjNZd6uFf9hbNC5RdfmHrzANBgkqhkiG9w0BAQsFADBb\\n    c8PH3PSoAaRwMMgOSA2ALJvbRz8mpg==\\n    -----END CERTIFICATE-----\\\"\\n    \"}"
+   },
+   "UpdateReplacePolicy": "Delete",
+   "DeletionPolicy": "Delete"
+  },
+  "ConfluentClientCertSecret23688D94": {
+   "Type": "AWS::SecretsManager::Secret",
+   "Properties": {
+    "SecretString": "{\"certificate\":\"-----BEGIN CERTIFICATE-----\\n    MIIE5DCCAsygAwIBAgIRAPJdwaFaNRrytHBto0j5BA0wDQYJKoZIhvcNAQELBQAw\\n    cmUuiAii9R0=\\n    -----END CERTIFICATE-----\\n    -----BEGIN CERTIFICATE-----\\n    MIIFgjCCA2qgAwIBAgIQdjNZd6uFf9hbNC5RdfmHrzANBgkqhkiG9w0BAQsFADBb\\n    c8PH3PSoAaRwMMgOSA2ALJvbRz8mpg==\\n    -----END CERTIFICATE-----\\\"\\n    \",\"privateKey\":\"-----BEGIN ENCRYPTED PRIVATE KEY-----\\n    zp2mwJn2NYB7AZ7+imp0azDZb+8YG2aUCiyqb6PnnA==\\n    -----END ENCRYPTED PRIVATE KEY-----\"}"
+   },
+   "UpdateReplacePolicy": "Delete",
+   "DeletionPolicy": "Delete"
+  }
+ },
+ "Parameters": {
+  "BootstrapVersion": {
+   "Type": "AWS::SSM::Parameter::Value<String>",
+   "Default": "/cdk-bootstrap/hnb659fds/version",
+   "Description": "Version of the CDK Bootstrap resources in this environment, automatically retrieved from SSM Parameter Store. [cdk:skip]"
+  }
+ },
+ "Rules": {
+  "CheckBootstrapVersion": {
+   "Assertions": [
+    {
+     "Assert": {
+      "Fn::Not": [
+       {
+        "Fn::Contains": [
+         [
+          "1",
+          "2",
+          "3",
+          "4",
+          "5"
+         ],
+         {
+          "Ref": "BootstrapVersion"
+         }
+        ]
+       }
+      ]
+     },
+     "AssertDescription": "CDK bootstrap stack version 6 required. Please run 'cdk bootstrap' with a recent version of the CDK CLI."
+    }
+   ]
+  }
+ }
+}