feat(lambda): support for schema registry for kafka

[DamFlancoz]

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

• My code adheres to the CONTRIBUTING GUIDE and DESIGN GUIDELINES

---

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

[kumvprat]

Can you regenerate the snapshots by setting this flag as false ?

This the default value of the flag it would be good to have that as a guidance for anyone working with these set of features in the future

[DamFlancoz]

Removing the flag added the destructive actions warning back. It also updated manifest to have those warnings:

"/lambda-event-source-glue-schema-registry/GlueFunction/inlinePolicyAddedToExecutionRole-0/Resource": [
          {
            "type": "aws:cdk:logicalId",
            "data": "GlueFunctioninlinePolicyAddedToExecutionRole043C5A5B3",
            "trace": [
              "!!DESTRUCTIVE_CHANGES: WILL_DESTROY"
            ]
          }
        ],

[kumvprat]

LGTM

[mergify]

Thank you for contributing! Your pull request will be updated from main and then merged automatically (do not update manually, and be sure to allow changes to be pushed to your fork).

[aws-cdk-automation]

AWS CodeBuild CI Report

• CodeBuild project: AutoBuildv2Project1C6BFA3F-wQm2hXv2jqQv
• Commit ID: 5d96c4f
• Result: SUCCEEDED
• Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository

[mergify]

Thank you for contributing! Your pull request will be updated from main and then merged automatically (do not update manually, and be sure to allow changes to be pushed to your fork).

[github-actions]

Comments on closed issues and PRs are hard for our team to see.
If you need help, please open a new issue that references this one.