diff --git a/.cfnlintrc.yaml b/.cfnlintrc.yaml index b704fa915..cee14a760 100644 --- a/.cfnlintrc.yaml +++ b/.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) diff --git a/samtranslator/__init__.py b/samtranslator/__init__.py index 1b10527ab..2876210d1 100644 --- a/samtranslator/__init__.py +++ b/samtranslator/__init__.py @@ -1 +1 @@ -__version__ = "1.115.0" +__version__ = "1.116.0" diff --git a/samtranslator/internal/schema_source/aws_serverless_function.py b/samtranslator/internal/schema_source/aws_serverless_function.py index 869993798..ad38e264c 100644 --- a/samtranslator/internal/schema_source/aws_serverless_function.py +++ b/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") diff --git a/samtranslator/internal/schema_source/sam-docs.json b/samtranslator/internal/schema_source/sam-docs.json index 3e8d28aff..695233fda 100644 --- a/samtranslator/internal/schema_source/sam-docs.json +++ b/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.", diff --git a/samtranslator/model/eventsources/pull.py b/samtranslator/model/eventsources/pull.py index 78b41700a..38a31f67a 100644 --- a/samtranslator/model/eventsources/pull.py +++ b/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( diff --git a/samtranslator/schema/schema.json b/samtranslator/schema/schema.json index af61d4b9f..f6c50ee7b 100644 --- a/samtranslator/schema/schema.json +++ b/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": [ { diff --git a/schema_source/sam.schema.json b/schema_source/sam.schema.json index 8930ddb79..03a2de9b1 100644 --- a/schema_source/sam.schema.json +++ b/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": [ { diff --git a/tests/model/eventsources/test_msk_event_source.py b/tests/model/eventsources/test_msk_event_source.py index 6163df28d..01b2c96e4 100644 --- a/tests/model/eventsources/test_msk_event_source.py +++ b/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)) diff --git a/tests/model/eventsources/test_self_managed_kafka_event_source.py b/tests/model/eventsources/test_self_managed_kafka_event_source.py index 779836c34..536639777 100644 --- a/tests/model/eventsources/test_self_managed_kafka_event_source.py +++ b/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"}, + ) diff --git a/tests/translator/input/error_consumption_mode_for_msk.yaml b/tests/translator/input/error_consumption_mode_for_msk.yaml new file mode 100644 index 000000000..5bd46a49d --- /dev/null +++ b/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 diff --git a/tests/translator/input/function_with_self_managed_kafka_consumption_mode.yaml b/tests/translator/input/function_with_self_managed_kafka_consumption_mode.yaml new file mode 100644 index 000000000..737437dd2 --- /dev/null +++ b/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 diff --git a/tests/translator/output/aws-cn/function_with_self_managed_kafka_consumption_mode.json b/tests/translator/output/aws-cn/function_with_self_managed_kafka_consumption_mode.json new file mode 100644 index 000000000..c47590ecb --- /dev/null +++ b/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" + } + } +} diff --git a/tests/translator/output/aws-us-gov/function_with_self_managed_kafka_consumption_mode.json b/tests/translator/output/aws-us-gov/function_with_self_managed_kafka_consumption_mode.json new file mode 100644 index 000000000..61dfe47f3 --- /dev/null +++ b/tests/translator/output/aws-us-gov/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-us-gov: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" + } + } +} diff --git a/tests/translator/output/error_consumption_mode_for_msk.json b/tests/translator/output/error_consumption_mode_for_msk.json new file mode 100644 index 000000000..e97ed0f50 --- /dev/null +++ b/tests/translator/output/error_consumption_mode_for_msk.json @@ -0,0 +1,10 @@ +{ + "_autoGeneratedBreakdownErrorMessage": [ + "Invalid Serverless Application Specification document. ", + "Number of errors found: 1. ", + "Resource with id [MSKFunctionWithConsumptionMode] is invalid. ", + "Event with id [MSKFunctionWithConsumptionModeMyKafkaCluster] is invalid. ", + "Property ConsumptionMode not defined for resource of type MSK." + ], + "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MSKFunctionWithConsumptionMode] is invalid. Event with id [MSKFunctionWithConsumptionModeMyKafkaCluster] is invalid. Property ConsumptionMode not defined for resource of type MSK." +} diff --git a/tests/translator/output/function_with_self_managed_kafka_consumption_mode.json b/tests/translator/output/function_with_self_managed_kafka_consumption_mode.json new file mode 100644 index 000000000..d5b11c7bd --- /dev/null +++ b/tests/translator/output/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: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" + } + } +}