Merge main to develop

.cfnlintrc.yaml

@@ -112,6 +112,7 @@ ignore_templates:
   - tests/translator/output/**/function_with_self_managed_kafka_iam_auth.json # cfnlint is not updated to recognize IAM_AUTH SourceAccessConfigurations type
   - tests/translator/output/**/function_with_self_managed_kafka_iam_oauth.json # cfnlint is not updated to recognize IAM_OAUTHBEARER_AUTH SourceAccessConfigurations type
   - tests/translator/output/**/function_with_msk_with_schema_registry_config.json # cfnlint is not updated to recognize the SchemaRegistryConfig property
+  - tests/translator/output/**/function_with_self_managed_kafka_consumption_mode.json # cfnlint is not updated to recognize the ConsumptionMode property
   - tests/translator/output/**/function_with_logging_config.json # cfnlint is not updated to recognize the LoggingConfig property
   - tests/translator/output/aws-*/*capacity_provider*.json # Ignore Capacity Provider test format in non-aws partitions
   - tests/translator/output/**/capacity_provider_managed_resource_tags.json # cfnlint not updated for CapacityProvider mode (Phase 2)

samtranslator/__init__.py

@@ -1 +1 @@
-__version__ = "1.115.0"
+__version__ = "1.116.0"

samtranslator/internal/schema_source/aws_serverless_function.py

@@ -494,6 +494,7 @@ class MQEvent(BaseModel):
 class SelfManagedKafkaEventProperties(BaseModel):
     BatchSize: PassThroughProp | None = selfmanagedkafkaeventproperties("BatchSize")
     ConsumerGroupId: PassThroughProp | None = selfmanagedkafkaeventproperties("ConsumerGroupId")
+    ConsumptionMode: PassThroughProp | None = selfmanagedkafkaeventproperties("ConsumptionMode")
     Enabled: PassThroughProp | None = selfmanagedkafkaeventproperties("Enabled")
     FilterCriteria: PassThroughProp | None = selfmanagedkafkaeventproperties("FilterCriteria")
     KafkaBootstrapServers: list[SamIntrinsicable[str]] | None = selfmanagedkafkaeventproperties("KafkaBootstrapServers")

samtranslator/internal/schema_source/sam-docs.json

@@ -426,6 +426,7 @@
     "sam-property-function-selfmanagedkafka": {
       "BatchSize": "The maximum number of records in each batch that Lambda pulls from your stream and sends to your function.  \n*Type*: Integer  \n*Required*: No  \n*Default*: 100  \n*CloudFormation compatibility*: This property is passed directly to the [`BatchSize`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-batchsize) property of an `AWS::Lambda::EventSourceMapping` resource.  \n*Minimum*: `1`  \n*Maximum*: `10000`",
       "ConsumerGroupId": "A string that configures how events will be read from Kafka topics.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`SelfManagedKafkaConfiguration`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html) property of an `AWS::Lambda::EventSourceMapping` resource.",
+      "ConsumptionMode": "The consumption mode of the event source mapping\\.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility:* This property is passed directly to the [`ConsumptionMode`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-selfmanagedkafkaeventsourceconfig.html) property of the `SelfManagedKafkaEventSourceConfig` data type of an `AWS::Lambda::EventSourceMapping` resource\\.",
       "Enabled": "Disables the event source mapping to pause polling and invocation.  \n*Type*: Boolean  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`Enabled`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-enabled) property of an `AWS::Lambda::EventSourceMapping` resource.",
       "FilterCriteria": "A object that defines the criteria to determine whether Lambda should process an event. For more information, see [AWS Lambda event filtering](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html) in the *AWS Lambda Developer Guide*.  \n*Type*: [FilterCriteria](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-filtercriteria.html)  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`FilterCriteria`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-filtercriteria.html) property of an `AWS::Lambda::EventSourceMapping` resource.",
       "KafkaBootstrapServers": "The list of bootstrap servers for your Kafka brokers. Include the port, for example `broker.example.com:xxxx`  \n*Type*: List  \n*Required*: No  \n*CloudFormation compatibility*: This property is unique to AWS SAM and doesn't have an CloudFormation equivalent.",

samtranslator/model/eventsources/pull.py

@@ -62,6 +62,7 @@ class PullEventSource(ResourceMacro, metaclass=ABCMeta):
         "SchemaRegistryConfig": PropertyType(False, IS_DICT),
         "MetricsConfig": PropertyType(False, IS_DICT),
         "LoggingConfig": PropertyType(False, IS_DICT),
+        "ConsumptionMode": PropertyType(False, IS_STR),
     }
 
     BatchSize: Intrinsicable[int] | None
@@ -89,6 +90,7 @@ class PullEventSource(ResourceMacro, metaclass=ABCMeta):
     SchemaRegistryConfig: dict[str, Any] | None
     MetricsConfig: dict[str, Any] | None
     LoggingConfig: dict[str, Any] | None
+    ConsumptionMode: Intrinsicable[str] | None
 
     @abstractmethod
     def get_policy_arn(self) -> str | None:
@@ -192,6 +194,17 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
                 lambda_eventsourcemapping.SelfManagedKafkaEventSourceConfig["SchemaRegistryConfig"] = (  # type: ignore[attr-defined]
                     self.SchemaRegistryConfig
                 )
+        if self.ConsumptionMode:
+            if self.resource_type != "SelfManagedKafka":
+                raise InvalidEventException(
+                    self.logical_id,
+                    f"Property ConsumptionMode not defined for resource of type {self.resource_type}.",
+                )
+            if not lambda_eventsourcemapping.SelfManagedKafkaEventSourceConfig:  # type: ignore[attr-defined]
+                lambda_eventsourcemapping.SelfManagedKafkaEventSourceConfig = {}
+            lambda_eventsourcemapping.SelfManagedKafkaEventSourceConfig["ConsumptionMode"] = (  # type: ignore[attr-defined]
+                self.ConsumptionMode
+            )
         destination_config_policy: dict[str, Any] | None = None
         if self.DestinationConfig:
             on_failure: dict[str, Any] = sam_expect(

samtranslator/schema/schema.json

@@ -441343,6 +441343,15 @@
           "markdownDescription": "A string that configures how events will be read from Kafka topics.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`SelfManagedKafkaConfiguration`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html) property of an `AWS::Lambda::EventSourceMapping` resource.",
           "title": "ConsumerGroupId"
         },
+        "ConsumptionMode": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/PassThroughProp"
+            }
+          ],
+          "markdownDescription": "The consumption mode of the event source mapping\\.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility:* This property is passed directly to the [`ConsumptionMode`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-selfmanagedkafkaeventsourceconfig.html) property of the `SelfManagedKafkaEventSourceConfig` data type of an `AWS::Lambda::EventSourceMapping` resource\\.",
+          "title": "ConsumptionMode"
+        },
         "Enabled": {
           "allOf": [
             {

schema_source/sam.schema.json

@@ -4088,6 +4088,15 @@
           "markdownDescription": "A string that configures how events will be read from Kafka topics.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility*: This property is passed directly to the [`SelfManagedKafkaConfiguration`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html) property of an `AWS::Lambda::EventSourceMapping` resource.",
           "title": "ConsumerGroupId"
         },
+        "ConsumptionMode": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/PassThroughProp"
+            }
+          ],
+          "markdownDescription": "The consumption mode of the event source mapping\\.  \n*Type*: String  \n*Required*: No  \n*CloudFormation compatibility:* This property is passed directly to the [`ConsumptionMode`](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-eventsourcemapping-selfmanagedkafkaeventsourceconfig.html) property of the `SelfManagedKafkaEventSourceConfig` data type of an `AWS::Lambda::EventSourceMapping` resource\\.",
+          "title": "ConsumptionMode"
+        },
         "Enabled": {
           "allOf": [
             {

tests/model/eventsources/test_msk_event_source.py

@@ -3,6 +3,7 @@
 
 from samtranslator.intrinsics.resolver import IntrinsicsResolver
 from samtranslator.model.eventsources.pull import MSK
+from samtranslator.model.exceptions import InvalidEventException
 from samtranslator.model.lambda_ import LambdaFunction
 
 
@@ -391,3 +392,17 @@ def test_to_cloudformation_with_metrics_config_multiple_metrics(self):
         self.assertEqual(len(resources), 1)
         event_source_mapping = resources[0]
         self.assertEqual(event_source_mapping.MetricsConfig, {"Metrics": ["EventCount", "ErrorCount", "KafkaMetrics"]})
+
+    def test_to_cloudformation_with_consumption_mode_raises(self):
+        self.kafka_event_source.Stream = "arn:aws:kafka:us-west-2:012345678901:cluster/mycluster/abc123"
+        self.kafka_event_source.Topics = ["MyTopic"]
+        self.kafka_event_source.SourceAccessConfigurations = [
+            {"Type": "CLIENT_CERTIFICATE_TLS_AUTH", "URI": "SECRET_URI"},
+        ]
+        self.kafka_event_source.ConsumptionMode = "Queue"
+
+        mock_function = LambdaFunction("TestFunction")
+        with self.assertRaises(InvalidEventException) as context:
+            self.kafka_event_source.to_cloudformation(function=mock_function)
+
+        self.assertIn("Property ConsumptionMode not defined for resource of type MSK.", str(context.exception))

tests/model/eventsources/test_self_managed_kafka_event_source.py

@@ -1126,3 +1126,33 @@ def test_to_cloudformation_with_metrics_config_multiple_metrics(self):
         self.assertEqual(len(resources), 1)
         event_source_mapping = resources[0]
         self.assertEqual(event_source_mapping.MetricsConfig, {"Metrics": ["EventCount", "ErrorCount", "KafkaMetrics"]})
+
+    def test_to_cloudformation_with_consumption_mode(self):
+        self.kafka_event_source.KafkaBootstrapServers = ["endpoint1", "endpoint2"]
+        self.kafka_event_source.Topics = ["test-topic"]
+        self.kafka_event_source.SourceAccessConfigurations = [{"Type": "BASIC_AUTH", "URI": "SECRET_URI"}]
+        self.kafka_event_source.ConsumptionMode = "Queue"
+
+        mock_function = LambdaFunction("TestFunction")
+        resources = self.kafka_event_source.to_cloudformation(function=mock_function)
+
+        self.assertEqual(len(resources), 1)
+        event_source_mapping = resources[0]
+        self.assertEqual(event_source_mapping.SelfManagedKafkaEventSourceConfig, {"ConsumptionMode": "Queue"})
+
+    def test_to_cloudformation_with_consumption_mode_and_consumer_group_id(self):
+        self.kafka_event_source.KafkaBootstrapServers = ["endpoint1", "endpoint2"]
+        self.kafka_event_source.Topics = ["test-topic"]
+        self.kafka_event_source.SourceAccessConfigurations = [{"Type": "BASIC_AUTH", "URI": "SECRET_URI"}]
+        self.kafka_event_source.ConsumerGroupId = "consumergroup"
+        self.kafka_event_source.ConsumptionMode = "Queue"
+
+        mock_function = LambdaFunction("TestFunction")
+        resources = self.kafka_event_source.to_cloudformation(function=mock_function)
+
+        self.assertEqual(len(resources), 1)
+        event_source_mapping = resources[0]
+        self.assertEqual(
+            event_source_mapping.SelfManagedKafkaEventSourceConfig,
+            {"ConsumerGroupId": "consumergroup", "ConsumptionMode": "Queue"},
+        )

tests/translator/input/error_consumption_mode_for_msk.yaml

@@ -0,0 +1,21 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Parameters: {}
+Resources:
+  MSKFunctionWithConsumptionMode:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/kafka.zip
+      Handler: index.kafka_handler
+      Runtime: python3.9
+      Events:
+        MyKafkaCluster:
+          Type: MSK
+          Properties:
+            Stream: !Sub arn:aws:kafka:${AWS::Region}:012345678901:cluster/mycluster/6cc0432b-8618-4f44-bccc-e1fbd8fb7c4d-2
+            StartingPosition: LATEST
+            SourceAccessConfigurations:
+            - Type: SASL_SCRAM_512_AUTH
+              URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+            Topics:
+            - abcd
+            ConsumptionMode: Queue

tests/translator/input/function_with_self_managed_kafka_consumption_mode.yaml

@@ -0,0 +1,29 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Parameters: {}
+Resources:
+  KafkaFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/kafka.zip
+      Handler: index.kafka_handler
+      Runtime: python3.9
+      Events:
+        MyKafkaCluster:
+          Type: SelfManagedKafka
+          Properties:
+            KafkaBootstrapServers:
+            - abc.xyz.com:9092
+            - 123.45.67.89:9096
+            Topics:
+            - Topic1
+            SourceAccessConfigurations:
+            - Type: SASL_SCRAM_512_AUTH
+              URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+            - Type: VPC_SUBNET
+              URI: subnet:subnet-12345
+            - Type: VPC_SECURITY_GROUP
+              URI: security_group:sg-67890
+            ConsumerGroupId: consumergroup1
+            ConsumptionMode: Queue
+            StartingPosition: AT_TIMESTAMP
+            StartingPositionTimestamp: 1672560000

tests/translator/output/aws-cn/function_with_self_managed_kafka_consumption_mode.json

@@ -0,0 +1,127 @@
+{
+  "AWSTemplateFormatVersion": "2010-09-09",
+  "Parameters": {},
+  "Resources": {
+    "KafkaFunction": {
+      "Properties": {
+        "Code": {
+          "S3Bucket": "sam-demo-bucket",
+          "S3Key": "kafka.zip"
+        },
+        "Handler": "index.kafka_handler",
+        "Role": {
+          "Fn::GetAtt": [
+            "KafkaFunctionRole",
+            "Arn"
+          ]
+        },
+        "Runtime": "python3.9",
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      },
+      "Type": "AWS::Lambda::Function"
+    },
+    "KafkaFunctionMyKafkaCluster": {
+      "Properties": {
+        "FunctionName": {
+          "Ref": "KafkaFunction"
+        },
+        "SelfManagedEventSource": {
+          "Endpoints": {
+            "KafkaBootstrapServers": [
+              "abc.xyz.com:9092",
+              "123.45.67.89:9096"
+            ]
+          }
+        },
+        "SelfManagedKafkaEventSourceConfig": {
+          "ConsumerGroupId": "consumergroup1",
+          "ConsumptionMode": "Queue"
+        },
+        "SourceAccessConfigurations": [
+          {
+            "Type": "SASL_SCRAM_512_AUTH",
+            "URI": "arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c"
+          },
+          {
+            "Type": "VPC_SUBNET",
+            "URI": "subnet:subnet-12345"
+          },
+          {
+            "Type": "VPC_SECURITY_GROUP",
+            "URI": "security_group:sg-67890"
+          }
+        ],
+        "StartingPosition": "AT_TIMESTAMP",
+        "StartingPositionTimestamp": 1672560000,
+        "Topics": [
+          "Topic1"
+        ]
+      },
+      "Type": "AWS::Lambda::EventSourceMapping"
+    },
+    "KafkaFunctionRole": {
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "sts:AssumeRole"
+              ],
+              "Effect": "Allow",
+              "Principal": {
+                "Service": [
+                  "lambda.amazonaws.com"
+                ]
+              }
+            }
+          ],
+          "Version": "2012-10-17"
+        },
+        "ManagedPolicyArns": [
+          "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+        ],
+        "Policies": [
+          {
+            "PolicyDocument": {
+              "Statement": [
+                {
+                  "Action": [
+                    "secretsmanager:GetSecretValue"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": "arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c"
+                },
+                {
+                  "Action": [
+                    "ec2:CreateNetworkInterface",
+                    "ec2:DescribeNetworkInterfaces",
+                    "ec2:DeleteNetworkInterface",
+                    "ec2:DescribeVpcs",
+                    "ec2:DescribeSubnets",
+                    "ec2:DescribeSecurityGroups"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": "*"
+                }
+              ],
+              "Version": "2012-10-17"
+            },
+            "PolicyName": "SelfManagedKafkaExecutionRolePolicy"
+          }
+        ],
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      },
+      "Type": "AWS::IAM::Role"
+    }
+  }
+}