Add support for KafkaSchemaRegistry for MSK and SMK

Author: vicheey

.cfnlintrc.yaml

@@ -139,6 +139,8 @@ ignore_templates:
   - tests/translator/output/**/managed_policies_everything.json # intentionally contains wrong arns
   - tests/translator/output/**/function_with_provisioned_poller_config.json
   - tests/translator/output/**/function_with_metrics_config.json
+  - tests/translator/output/**/function_with_self_managed_kafka_and_schema_registry.json # cfnlint is not updated to recognize the SchemaRegistryConfig property
+  - tests/translator/output/**/function_with_msk_with_schema_registry_config.json # cfnlint is not updated to recognize the SchemaRegistryConfig property
 
 ignore_checks:
   - E2531 # Deprecated runtime; not relevant for transform tests

integration/combination/test_function_with_msk.py

@@ -41,6 +41,15 @@ def test_function_with_msk_trigger_and_s3_onfailure_events_destinations(self):
             "combination/function_with_msk_trigger_and_s3_onfailure_events_destinations", parameters
         )
 
+    def test_function_with_msk_trigger_and_confluent_schema_registry(self):
+        companion_stack_outputs = self.companion_stack_outputs
+        parameters = self.get_parameters(companion_stack_outputs)
+        cluster_name = "MskCluster4-" + generate_suffix()
+        parameters.append(self.generate_parameter("MskClusterName4", cluster_name))
+        self._common_validations_for_MSK(
+            "combination/function_with_msk_trigger_and_confluent_schema_registry", parameters
+        )
+
     def _common_validations_for_MSK(self, file_name, parameters):
         self.create_and_verify_stack(file_name, parameters)
 

integration/resources/expected/combination/function_with_msk_trigger_and_confluent_schema_registry.json

@@ -0,0 +1,22 @@
+[
+  {
+    "LogicalResourceId": "MyMskStreamProcessor",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "MyLambdaExecutionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "MyMskCluster",
+    "ResourceType": "AWS::MSK::Cluster"
+  },
+  {
+    "LogicalResourceId": "MyMskStreamProcessorMyMskEvent",
+    "ResourceType": "AWS::Lambda::EventSourceMapping"
+  },
+  {
+    "LogicalResourceId": "PreCreatedS3Bucket",
+    "ResourceType": "AWS::S3::Bucket"
+  }
+]

integration/resources/templates/combination/function_with_msk.yaml

@@ -45,7 +45,7 @@ Resources:
             VolumeSize: 1
       ClusterName:
         Ref: MskClusterName
-      KafkaVersion: 2.4.1.1
+      KafkaVersion: 3.8.x
       NumberOfBrokerNodes: 2
 
   MyMskStreamProcessor:

integration/resources/templates/combination/function_with_msk_trigger_and_confluent_schema_registry.yaml

@@ -0,0 +1,86 @@
+Parameters:
+  PreCreatedSubnetOne:
+    Type: String
+  PreCreatedSubnetTwo:
+    Type: String
+  MskClusterName4:
+    Type: String
+
+Resources:
+  MyLambdaExecutionRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+        - Action: [sts:AssumeRole]
+          Effect: Allow
+          Principal:
+            Service: [lambda.amazonaws.com]
+      Policies:
+      - PolicyName: IntegrationTestExecution
+        PolicyDocument:
+          Statement:
+          - Action: [kafka:DescribeCluster, kafka:GetBootstrapBrokers, ec2:CreateNetworkInterface,
+              ec2:DescribeNetworkInterfaces, ec2:DescribeVpcs, ec2:DeleteNetworkInterface,
+              ec2:DescribeSubnets, ec2:DescribeSecurityGroups, logs:CreateLogGroup,
+              logs:CreateLogStream, logs:PutLogEvents, s3:ListBucket, kafka:DescribeClusterV2]
+            Effect: Allow
+            Resource: '*'
+      ManagedPolicyArns:
+      - !Sub arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
+      Tags:
+      - {Value: SAM, Key: lambda:createdBy}
+
+  MyMskCluster:
+    Type: AWS::MSK::Cluster
+    Properties:
+      BrokerNodeGroupInfo:
+        ClientSubnets:
+        - Ref: PreCreatedSubnetOne
+        - Ref: PreCreatedSubnetTwo
+        InstanceType: kafka.t3.small
+        StorageInfo:
+          EBSStorageInfo:
+            VolumeSize: 1
+      ClusterName:
+        Ref: MskClusterName4
+      KafkaVersion: 3.8.x
+      NumberOfBrokerNodes: 2
+
+  MyMskStreamProcessor:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs18.x
+      Handler: index.handler
+      CodeUri: ${codeuri}
+      Role:
+        Fn::GetAtt: [MyLambdaExecutionRole, Arn]
+      Events:
+        MyMskEvent:
+          Type: MSK
+          Properties:
+            StartingPosition: LATEST
+            Stream:
+              Ref: MyMskCluster
+            Topics:
+            - SchemaRegistryTestTopic
+            ProvisionedPollerConfig:
+              MinimumPollers: 1
+            SchemaRegistryConfig:
+              AccessConfigs:
+              - Type: BASIC_AUTH
+                URI: !Sub arn:${AWS::Partition}:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+              SchemaValidationConfigs:
+              - Attribute: KEY
+              EventRecordFormat: JSON
+              SchemaRegistryURI: https://confluent.us-east-2.aws.confluent.cloud:9092
+
+
+
+  PreCreatedS3Bucket:
+    Type: AWS::S3::Bucket
+    DeletionPolicy: Delete
+
+Metadata:
+  SamTransformTest: true

integration/resources/templates/combination/function_with_msk_trigger_and_s3_onfailure_events_destinations.yaml

@@ -45,7 +45,7 @@ Resources:
             VolumeSize: 1
       ClusterName:
         Ref: MskClusterName3
-      KafkaVersion: 2.4.1.1
+      KafkaVersion: 3.8.x
       NumberOfBrokerNodes: 2
 
   MyMskStreamProcessor:

integration/resources/templates/combination/function_with_msk_using_managed_policy.yaml

@@ -20,7 +20,7 @@ Resources:
             VolumeSize: 1
       ClusterName:
         Ref: MskClusterName2
-      KafkaVersion: 2.4.1.1
+      KafkaVersion: 3.8.x
       NumberOfBrokerNodes: 2
 
   MyMskStreamProcessor:

requirements/base.txt

@@ -1,4 +1,4 @@
-boto3>=1.19.5,==1.*
+boto3>=1.34.0,<2.0.0
 jsonschema<5,>=3.2  # TODO: evaluate risk of removing jsonschema 3.x support
 typing_extensions>=4.4 # 3.8 doesn't have Required, TypeGuard and ParamSpec
 

requirements/dev.txt

@@ -12,8 +12,8 @@ parameterized~=0.7
 
 # Integration tests
 dateparser~=1.1
-boto3>=1.23,<2
-tenacity~=8.0
+boto3>=1.34.0,<2.0.0
+tenacity~=9.0
 
 # Requirements for examples
 requests~=2.28
@@ -26,6 +26,9 @@ ruamel.yaml==0.17.21  # It can parse yaml while perserving comments
 mypy~=1.3.0
 
 # types
-boto3-stubs[appconfig,serverlessrepo]>=1.19.5,==1.*
+boto3-stubs[appconfig,serverlessrepo]>=1.34.0,<2.0.0
 types-PyYAML~=6.0
 types-jsonschema~=3.2
+
+# CloudFormation CLI tools
+cloudformation-cli>=0.2.39,<0.3.0

samtranslator/internal/schema_source/aws_serverless_function.py

@@ -421,6 +421,7 @@ class MSKEventProperties(BaseModel):
     SourceAccessConfigurations: Optional[PassThroughProp] = mskeventproperties("SourceAccessConfigurations")
     DestinationConfig: Optional[PassThroughProp]  # TODO: add documentation
     ProvisionedPollerConfig: Optional[PassThroughProp]
+    SchemaRegistryConfig: Optional[PassThroughProp]
 
 
 class MSKEvent(BaseModel):
@@ -460,6 +461,7 @@ class SelfManagedKafkaEventProperties(BaseModel):
     StartingPositionTimestamp: Optional[PassThroughProp]  # TODO: add documentation
     Topics: PassThroughProp = selfmanagedkafkaeventproperties("Topics")
     ProvisionedPollerConfig: Optional[PassThroughProp]
+    SchemaRegistryConfig: Optional[PassThroughProp]
 
 
 class SelfManagedKafkaEvent(BaseModel):