Repository: getmoto/moto
Pull request #7384 (2024-02-22 23:07:18)

CF - Create-support for AWS::EMR::Cluster
See #7381 


Patch:
diff --git a/docs/docs/services/cf.rst b/docs/docs/services/cf.rst
index b3e227189..86d7797bc 100644
--- a/docs/docs/services/cf.rst
+++ b/docs/docs/services/cf.rst
@@ -121,6 +121,8 @@ Please let us know if you'd like support for a resource not yet listed here.
   +---------------------------------------+--------+--------+--------+ - [ ] Id                               |
   |                                       |        |        |        |                                        |
   +---------------------------------------+--------+--------+--------+----------------------------------------+
+  |AWS::EMR::Cluster                      |    x   |        |        | - [ ] MasterPublicDNS                  |
+  +---------------------------------------+--------+--------+--------+----------------------------------------+
   |AWS::Events::Archive                   |    x   |    x   |        | - [x] Arn                              |
   +---------------------------------------+--------+--------+--------+----------------------------------------+
   |AWS::Events::EventBus                  |    x   |    x   |    x   | - [x] Arn                              |
diff --git a/moto/autoscaling/models.py b/moto/autoscaling/models.py
index e835e71fa..2f27e3610 100644
--- a/moto/autoscaling/models.py
+++ b/moto/autoscaling/models.py
@@ -1190,13 +1190,40 @@ class AutoScalingBackend(BaseBackend):
         return group
 
     def describe_auto_scaling_groups(
-        self, names: List[str]
+        self, names: List[str], filters: Optional[List[Dict[str, str]]] = None
     ) -> List[FakeAutoScalingGroup]:
-        groups = self.autoscaling_groups.values()
+
+        groups = list(self.autoscaling_groups.values())
+
+        if filters:
+            for f in filters:
+                if f["Name"] == "tag-key":
+                    groups = [
+                        group
+                        for group in groups
+                        if any(tag["Key"] in f["Values"] for tag in group.tags)
+                    ]
+                elif f["Name"] == "tag-value":
+                    groups = [
+                        group
+                        for group in groups
+                        if any(tag["Value"] in f["Values"] for tag in group.tags)
+                    ]
+                elif f["Name"].startswith("tag:"):
+                    tag_key = f["Name"][4:]
+                    groups = [
+                        group
+                        for group in groups
+                        if any(
+                            tag["Key"] == tag_key and tag["Value"] in f["Values"]
+                            for tag in group.tags
+                        )
+                    ]
+
         if names:
-            return [group for group in groups if group.name in names]
-        else:
-            return list(groups)
+            groups = [group for group in groups if group.name in names]
+
+        return groups
 
     def delete_auto_scaling_group(self, group_name: str) -> None:
         self.set_desired_capacity(group_name, 0)
diff --git a/moto/autoscaling/responses.py b/moto/autoscaling/responses.py
index d83ca039e..e9aa149ca 100644
--- a/moto/autoscaling/responses.py
+++ b/moto/autoscaling/responses.py
@@ -228,7 +228,10 @@ class AutoScalingResponse(BaseResponse):
     def describe_auto_scaling_groups(self) -> str:
         names = self._get_multi_param("AutoScalingGroupNames.member")
         token = self._get_param("NextToken")
-        all_groups = self.autoscaling_backend.describe_auto_scaling_groups(names)
+        filters = self._get_params().get("Filters", [])
+        all_groups = self.autoscaling_backend.describe_auto_scaling_groups(
+            names, filters=filters
+        )
         all_names = [group.name for group in all_groups]
         if token:
             start = all_names.index(token) + 1
diff --git a/moto/cloudformation/parsing.py b/moto/cloudformation/parsing.py
index 8b0eaa608..239cc886d 100644
--- a/moto/cloudformation/parsing.py
+++ b/moto/cloudformation/parsing.py
@@ -33,8 +33,6 @@ from moto.awslambda import models as lambda_models  # noqa  # pylint: disable=al
 from moto.batch import models as batch_models  # noqa  # pylint: disable=all
 from moto.cloudformation.custom_model import CustomModel
 from moto.cloudwatch import models as cw_models  # noqa  # pylint: disable=all
-
-# End ugly list of imports
 from moto.core.common_models import CloudFormationModel
 from moto.datapipeline import models as data_models  # noqa  # pylint: disable=all
 from moto.dynamodb import models as ddb_models  # noqa  # pylint: disable=all
@@ -45,6 +43,7 @@ from moto.ecs import models as ecs_models  # noqa  # pylint: disable=all
 from moto.efs import models as efs_models  # noqa  # pylint: disable=all
 from moto.elb import models as elb_models  # noqa  # pylint: disable=all
 from moto.elbv2 import models as elbv2_models  # noqa  # pylint: disable=all
+from moto.emr import models as emr_models  # noqa  # pylint: disable=all
 from moto.events import models as events_models  # noqa  # pylint: disable=all
 from moto.iam import models as iam_models  # noqa  # pylint: disable=all
 from moto.kinesis import models as kinesis_models  # noqa  # pylint: disable=all
@@ -62,6 +61,7 @@ from moto.ssm import models as ssm_models  # noqa  # pylint: disable=all
 from moto.ssm import ssm_backends
 from moto.stepfunctions import models as sfn_models  # noqa  # pylint: disable=all
 
+# End ugly list of imports
 from .exceptions import (
     ExportNotFound,
     MissingParameterError,
diff --git a/moto/emr/models.py b/moto/emr/models.py
index fa0d7100e..58f65613c 100644
--- a/moto/emr/models.py
+++ b/moto/emr/models.py
@@ -4,7 +4,7 @@ from typing import Any, Dict, List, Optional, Tuple
 from dateutil.parser import parse as dtparse
 
 from moto.core.base_backend import BackendDict, BaseBackend
-from moto.core.common_models import BaseModel
+from moto.core.common_models import BaseModel, CloudFormationModel
 from moto.emr.exceptions import (
     InvalidRequestException,
     ResourceNotFoundException,
@@ -151,7 +151,7 @@ class FakeStep(BaseModel):
         self.start_datetime = datetime.now(timezone.utc)
 
 
-class FakeCluster(BaseModel):
+class FakeCluster(CloudFormationModel):
     def __init__(
         self,
         emr_backend: "ElasticMapReduceBackend",
@@ -384,6 +384,71 @@ class FakeCluster(BaseModel):
     def set_visibility(self, visibility: str) -> None:
         self.visible_to_all_users = visibility
 
+    @property
+    def physical_resource_id(self) -> str:
+        return self.id
+
+    @staticmethod
+    def cloudformation_type() -> str:
+        return "AWS::EMR::Cluster"
+
+    @classmethod
+    def has_cfn_attr(cls, attr: str) -> bool:
+        return attr in ["Id"]
+
+    def get_cfn_attribute(self, attribute_name: str) -> str:
+        from moto.cloudformation.exceptions import UnformattedGetAttTemplateException
+
+        if attribute_name == "Id":
+            return self.id
+        raise UnformattedGetAttTemplateException()
+
+    @classmethod
+    def create_from_cloudformation_json(  # type: ignore[misc]
+        cls,
+        resource_name: str,
+        cloudformation_json: Any,
+        account_id: str,
+        region_name: str,
+        **kwargs: Any,
+    ) -> "FakeCluster":
+
+        properties = cloudformation_json["Properties"]
+
+        instance_attrs = properties.get("Instances", {})
+        instance_attrs["ec2_subnet_id"] = instance_attrs.get("Ec2SubnetId")
+        instance_attrs["emr_managed_master_security_group"] = instance_attrs.get(
+            "EmrManagedMasterSecurityGroup"
+        )
+        instance_attrs["emr_managed_slave_security_group"] = instance_attrs.get(
+            "EmrManagedSlaveSecurityGroup"
+        )
+        instance_attrs["service_access_security_group"] = instance_attrs.get(
+            "ServiceAccessSecurityGroup"
+        )
+        instance_attrs["additional_master_security_groups"] = instance_attrs.get(
+            "AdditionalMasterSecurityGroups", []
+        )
+        instance_attrs["additional_slave_security_groups"] = instance_attrs.get(
+            "AdditionalSlaveSecurityGroups", []
+        )
+
+        emr_backend: ElasticMapReduceBackend = emr_backends[account_id][region_name]
+        cluster = emr_backend.run_job_flow(
+            name=properties["Name"],
+            log_uri=properties.get("LogUri"),
+            job_flow_role=properties["JobFlowRole"],
+            service_role=properties["ServiceRole"],
+            steps=[],
+            instance_attrs=instance_attrs,
+            kerberos_attributes=properties.get("KerberosAttributes", {}),
+            release_label=properties.get("ReleaseLabel"),
+            custom_ami_id=properties.get("CustomAmiId"),
+        )
+        tags = {item["Key"]: item["Value"] for item in properties.get("Tags", [])}
+        cluster.add_tags(tags)
+        return cluster
+
 
 class FakeSecurityConfiguration(BaseModel):
     def __init__(self, name: str, security_configuration: str):
diff --git a/moto/kms/models.py b/moto/kms/models.py
index f567a9c58..d12d9834a 100644
--- a/moto/kms/models.py
+++ b/moto/kms/models.py
@@ -233,11 +233,11 @@ class Key(CloudFormationModel):
             policy=properties["KeyPolicy"],
             key_usage="ENCRYPT_DECRYPT",
             key_spec="SYMMETRIC_DEFAULT",
-            description=properties["Description"],
+            description=properties.get("Description"),
             tags=properties.get("Tags", []),
         )
-        key.key_rotation_status = properties["EnableKeyRotation"]
-        key.enabled = properties["Enabled"]
+        key.key_rotation_status = properties.get("EnableKeyRotation", False)
+        key.enabled = properties.get("Enabled", True)
 
         return key
 


Test patch:
diff --git a/tests/test_autoscaling/test_autoscaling_group_filters.py b/tests/test_autoscaling/test_autoscaling_group_filters.py
new file mode 100644
index 000000000..914c23238
--- /dev/null
+++ b/tests/test_autoscaling/test_autoscaling_group_filters.py
@@ -0,0 +1,116 @@
+import boto3
+
+from moto import mock_aws
+from tests import EXAMPLE_AMI_ID
+
+from .utils import setup_networking
+
+
+@mock_aws
+def test_describe_autoscaling_groups_filter_by_tag_key():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups(
+        Filters=[{"Name": "tag-key", "Values": ["test_key1"]}]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+
+    assert len(response["AutoScalingGroups"]) == 1
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert {
+        "Key": "test_key1",
+        "PropagateAtLaunch": False,
+        "ResourceId": "test_asg1",
+        "ResourceType": "auto-scaling-group",
+        "Value": "test_value1",
+    } in tags
+
+
+@mock_aws
+def test_describe_autoscaling_groups_filter_by_tag_value():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups(
+        Filters=[{"Name": "tag-value", "Values": ["test_value1"]}]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+
+    assert len(response["AutoScalingGroups"]) == 1
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert {
+        "Key": "test_key1",
+        "PropagateAtLaunch": False,
+        "ResourceId": "test_asg1",
+        "ResourceType": "auto-scaling-group",
+        "Value": "test_value1",
+    } in tags
+
+
+@mock_aws
+def test_describe_autoscaling_groups_filter_by_tag_key_value():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups(
+        Filters=[{"Name": "tag:test_key1", "Values": ["test_value1"]}]
+    )
+    group = response["AutoScalingGroups"][0]
+    tags = group["Tags"]
+
+    assert len(response["AutoScalingGroups"]) == 1
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert {
+        "Key": "test_key1",
+        "PropagateAtLaunch": False,
+        "ResourceId": "test_asg1",
+        "ResourceType": "auto-scaling-group",
+        "Value": "test_value1",
+    } in tags
+
+
+@mock_aws
+def test_describe_autoscaling_groups_no_filter():
+    subnet = setup_networking()["subnet1"]
+    client = boto3.client("autoscaling", region_name="us-east-1")
+    create_asgs(client, subnet)
+
+    response = client.describe_auto_scaling_groups()
+
+    assert len(response["AutoScalingGroups"]) == 2
+    assert response["AutoScalingGroups"][0]["AutoScalingGroupName"] == "test_asg1"
+    assert response["AutoScalingGroups"][1]["AutoScalingGroupName"] == "test_asg2"
+
+
+def create_asgs(client, subnet):
+    _ = client.create_launch_configuration(
+        LaunchConfigurationName="test_launch_configuration",
+        ImageId=EXAMPLE_AMI_ID,
+        InstanceType="t2.medium",
+    )
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg1",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        Tags=[
+            {"Key": "test_key1", "Value": "test_value1"},
+            {"Key": "test_key2", "Value": "test_value2"},
+        ],
+        VPCZoneIdentifier=subnet,
+    )
+    client.create_auto_scaling_group(
+        AutoScalingGroupName="test_asg2",
+        LaunchConfigurationName="test_launch_configuration",
+        MinSize=0,
+        MaxSize=20,
+        DesiredCapacity=5,
+        VPCZoneIdentifier=subnet,
+    )
diff --git a/tests/test_emr/test_emr_cloudformation.py b/tests/test_emr/test_emr_cloudformation.py
new file mode 100644
index 000000000..3bebfd22a
--- /dev/null
+++ b/tests/test_emr/test_emr_cloudformation.py
@@ -0,0 +1,620 @@
+import json
+
+import boto3
+
+from moto import mock_aws
+from tests import EXAMPLE_AMI_ID
+
+template = """{
+  "AWSTemplateFormatVersion": "2010-09-09",
+  "Description": "The AWS CloudFormation template for this Serverless application",
+  "Resources": {
+    "Cluster1": {
+        "Type" : "AWS::EMR::Cluster",
+        "Properties" : {
+          "Instances" : {
+              "CoreInstanceGroup": {
+                  "InstanceCount": 3,
+                  "InstanceType": "m3g",
+              }
+          },
+          "JobFlowRole" : "EMR_EC2_DefaultRole",
+          "Name" : "my cluster",
+          "ServiceRole" : "EMR_DefaultRole",
+        }
+    },
+  },
+  "Outputs": {
+    "ClusterId": {
+        "Description": "Cluster info",
+        "Value": {
+            "Fn::GetAtt": ["Cluster1", "Id"]
+        },
+    }
+  }
+}"""
+
+
+@mock_aws
+def test_create_simple_cluster__using_cloudformation():
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(StackName="teststack", TemplateBody=template)
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"][0]
+    cluster_id = res["PhysicalResourceId"]
+    assert res["LogicalResourceId"] == "Cluster1"
+    assert res["ResourceType"] == "AWS::EMR::Cluster"
+    assert cluster_id.startswith("j-")
+
+    # Verify outputs
+    stack = cf.describe_stacks(StackName="teststack")["Stacks"][0]
+    assert {"OutputKey": "ClusterId", "OutputValue": cluster_id} in stack["Outputs"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "my cluster"
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == "EMR_EC2_DefaultRole"
+    assert cl["ServiceRole"] == "EMR_DefaultRole"
+    assert cl["Tags"] == []
+
+
+template_with_tags = {
+    "Resources": {
+        "Cluster1": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "Instances": {
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 3,
+                        "InstanceType": "m3g",
+                    }
+                },
+                "JobFlowRole": "EMR_EC2_DefaultRole",
+                "Name": "my cluster",
+                "ServiceRole": "EMR_DefaultRole",
+                "Tags": [
+                    {"Key": "k1", "Value": "v1"},
+                    {"Key": "k2", "Value": "v2"},
+                ],
+            },
+        },
+    },
+}
+
+
+@mock_aws
+def test_create_simple_cluster_with_tags():
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(StackName="teststack", TemplateBody=json.dumps(template_with_tags))
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"][0]
+    cluster_id = res["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Tags"] == [{"Key": "k1", "Value": "v1"}, {"Key": "k2", "Value": "v2"}]
+
+
+template_with_custom_ami = {
+    "AWSTemplateFormatVersion": "2010-09-09",
+    "Parameters": {
+        "SubnetId": {"Type": "String"},
+        "InstanceType": {"Type": "String"},
+        "TerminationProtected": {"Type": "String", "Default": "false"},
+        "ElasticMapReducePrincipal": {"Type": "String"},
+        "Ec2Principal": {"Type": "String"},
+    },
+    "Resources": {
+        "cluster": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "CustomAmiId": EXAMPLE_AMI_ID,
+                "Instances": {
+                    "MasterInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnMaster",
+                    },
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnCore",
+                    },
+                    "TaskInstanceGroups": [
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-1",
+                        },
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-2",
+                        },
+                    ],
+                    "TerminationProtected": {"Ref": "TerminationProtected"},
+                    "Ec2SubnetId": {"Ref": "SubnetId"},
+                },
+                "Name": "CFNtest",
+                "JobFlowRole": {"Ref": "emrEc2InstanceProfile"},
+                "ServiceRole": {"Ref": "emrRole"},
+                "ReleaseLabel": "release_2024",
+                "VisibleToAllUsers": True,
+                "Tags": [{"Key": "key1", "Value": "value1"}],
+            },
+        },
+        "emrRole": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {
+                                "Service": {"Ref": "ElasticMapReducePrincipal"}
+                            },
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+                ],
+            },
+        },
+        "emrEc2Role": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": {"Ref": "Ec2Principal"}},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role"
+                ],
+            },
+        },
+        "emrEc2InstanceProfile": {
+            "Type": "AWS::IAM::InstanceProfile",
+            "Properties": {"Path": "/", "Roles": [{"Ref": "emrEc2Role"}]},
+        },
+    },
+}
+
+
+@mock_aws
+def test_create_cluster_with_custom_ami():
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+
+    params = [
+        {"ParameterKey": "InstanceType", "ParameterValue": "i4s"},
+        {"ParameterKey": "SubnetId", "ParameterValue": subnet.id},
+        {
+            "ParameterKey": "ElasticMapReducePrincipal",
+            "ParameterValue": "emr_principal",
+        },
+        {"ParameterKey": "Ec2Principal", "ParameterValue": "ec2_principal"},
+    ]
+
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(
+        StackName="teststack",
+        TemplateBody=json.dumps(template_with_custom_ami),
+        Parameters=params,
+    )
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"]
+    cluster = [r for r in res if r["ResourceType"] == "AWS::EMR::Cluster"][0]
+    cluster_id = cluster["PhysicalResourceId"]
+
+    instance_profile = [
+        r for r in res if r["ResourceType"] == "AWS::IAM::InstanceProfile"
+    ][0]
+    instance_profile_id = instance_profile["PhysicalResourceId"]
+
+    role = [r for r in res if r["ResourceType"] == "AWS::IAM::Role"][0]
+    role_id = role["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "CFNtest"
+    assert cl["Ec2InstanceAttributes"]["Ec2SubnetId"] == subnet.id
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == instance_profile_id
+    assert cl["ServiceRole"] == role_id
+    assert cl["ReleaseLabel"] == "release_2024"
+    assert cl["CustomAmiId"] == EXAMPLE_AMI_ID
+
+
+template_with_root_volume = {
+    "AWSTemplateFormatVersion": "2010-09-09",
+    "Parameters": {
+        "InstanceType": {"Type": "String"},
+        "ReleaseLabel": {"Type": "String"},
+        "SubnetId": {"Type": "String"},
+        "TerminationProtected": {"Type": "String", "Default": "false"},
+        "EbsRootVolumeSize": {"Type": "String"},
+    },
+    "Resources": {
+        "cluster": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "EbsRootVolumeSize": {"Ref": "EbsRootVolumeSize"},
+                "Instances": {
+                    "MasterInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnMaster",
+                    },
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnCore",
+                    },
+                    "TaskInstanceGroups": [
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-1",
+                        },
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-2",
+                        },
+                    ],
+                    "TerminationProtected": {"Ref": "TerminationProtected"},
+                    "Ec2SubnetId": {"Ref": "SubnetId"},
+                },
+                "Name": "CFNtest",
+                "JobFlowRole": {"Ref": "emrEc2InstanceProfile"},
+                "ServiceRole": {"Ref": "emrRole"},
+                "ReleaseLabel": {"Ref": "ReleaseLabel"},
+                "VisibleToAllUsers": True,
+                "Tags": [{"Key": "key1", "Value": "value1"}],
+            },
+        },
+        "emrRole": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "elasticmapreduce.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+                ],
+            },
+        },
+        "emrEc2Role": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "ec2.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role"
+                ],
+            },
+        },
+        "emrEc2InstanceProfile": {
+            "Type": "AWS::IAM::InstanceProfile",
+            "Properties": {"Path": "/", "Roles": [{"Ref": "emrEc2Role"}]},
+        },
+    },
+}
+
+
+@mock_aws
+def test_create_cluster_with_root_volume():
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+
+    params = [
+        {"ParameterKey": "InstanceType", "ParameterValue": "i4s"},
+        {"ParameterKey": "ReleaseLabel", "ParameterValue": "latest"},
+        {"ParameterKey": "EbsRootVolumeSize", "ParameterValue": "15"},
+        {"ParameterKey": "SubnetId", "ParameterValue": subnet.id},
+        {
+            "ParameterKey": "ElasticMapReducePrincipal",
+            "ParameterValue": "emr_principal",
+        },
+        {"ParameterKey": "Ec2Principal", "ParameterValue": "ec2_principal"},
+    ]
+
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(
+        StackName="teststack",
+        TemplateBody=json.dumps(template_with_root_volume),
+        Parameters=params,
+    )
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"]
+    cluster = [r for r in res if r["ResourceType"] == "AWS::EMR::Cluster"][0]
+    cluster_id = cluster["PhysicalResourceId"]
+
+    instance_profile = [
+        r for r in res if r["ResourceType"] == "AWS::IAM::InstanceProfile"
+    ][0]
+    instance_profile_id = instance_profile["PhysicalResourceId"]
+
+    role = [r for r in res if r["ResourceType"] == "AWS::IAM::Role"][0]
+    role_id = role["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "CFNtest"
+    assert cl["Ec2InstanceAttributes"]["Ec2SubnetId"] == subnet.id
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == instance_profile_id
+    assert cl["ServiceRole"] == role_id
+
+
+template_with_kerberos_attrs = {
+    "AWSTemplateFormatVersion": "2010-09-09",
+    "Parameters": {
+        "CrossRealmTrustPrincipalPassword": {"Type": "String"},
+        "KdcAdminPassword": {"Type": "String"},
+        "InstanceType": {"Type": "String"},
+        "ReleaseLabel": {"Type": "String"},
+        "SubnetId": {"Type": "String"},
+    },
+    "Resources": {
+        "cluster": {
+            "Type": "AWS::EMR::Cluster",
+            "Properties": {
+                "Instances": {
+                    "MasterInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnMaster",
+                    },
+                    "CoreInstanceGroup": {
+                        "InstanceCount": 1,
+                        "InstanceType": {"Ref": "InstanceType"},
+                        "Market": "ON_DEMAND",
+                        "Name": "cfnCore",
+                    },
+                    "TaskInstanceGroups": [
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-1",
+                        },
+                        {
+                            "InstanceCount": 1,
+                            "InstanceType": {"Ref": "InstanceType"},
+                            "Market": "ON_DEMAND",
+                            "Name": "cfnTask-2",
+                        },
+                    ],
+                    "Ec2SubnetId": {"Ref": "SubnetId"},
+                },
+                "Name": "CFNtest",
+                "JobFlowRole": {"Ref": "emrEc2InstanceProfile"},
+                "KerberosAttributes": {
+                    "CrossRealmTrustPrincipalPassword": {
+                        "Ref": "CrossRealmTrustPrincipalPassword"
+                    },
+                    "KdcAdminPassword": {"Ref": "KdcAdminPassword"},
+                    "Realm": "EC2.INTERNAL",
+                },
+                "ServiceRole": {"Ref": "emrRole"},
+                "ReleaseLabel": {"Ref": "ReleaseLabel"},
+                "SecurityConfiguration": {"Ref": "securityConfiguration"},
+                "VisibleToAllUsers": True,
+                "Tags": [{"Key": "key1", "Value": "value1"}],
+            },
+        },
+        "key": {
+            "Type": "AWS::KMS::Key",
+            "Properties": {
+                "KeyPolicy": {
+                    "Version": "2012-10-17",
+                    "Id": "key-default-1",
+                    "Statement": [
+                        {
+                            "Sid": "Enable IAM User Permissions",
+                            "Effect": "Allow",
+                            "Principal": {"AWS": {"Fn::GetAtt": ["emrEc2Role", "Arn"]}},
+                            "Action": "kms:*",
+                            "Resource": "*",
+                        },
+                        {
+                            "Sid": "Enable IAM User Permissions",
+                            "Effect": "Allow",
+                            "Principal": {
+                                "AWS": {
+                                    "Fn::Join": [
+                                        "",
+                                        [
+                                            "arn:aws:iam::",
+                                            {"Ref": "AWS::AccountId"},
+                                            ":root",
+                                        ],
+                                    ]
+                                }
+                            },
+                            "Action": "kms:*",
+                            "Resource": "*",
+                        },
+                    ],
+                }
+            },
+        },
+        "securityConfiguration": {
+            "Type": "AWS::EMR::SecurityConfiguration",
+            "Properties": {
+                "SecurityConfiguration": {
+                    "AuthenticationConfiguration": {
+                        "KerberosConfiguration": {
+                            "Provider": "ClusterDedicatedKdc",
+                            "ClusterDedicatedKdcConfiguration": {
+                                "TicketLifetimeInHours": 24,
+                                "CrossRealmTrustConfiguration": {
+                                    "Realm": "AD.DOMAIN.COM",
+                                    "Domain": "ad.domain.com",
+                                    "AdminServer": "ad.domain.com",
+                                    "KdcServer": "ad.domain.com",
+                                },
+                            },
+                        }
+                    }
+                }
+            },
+        },
+        "emrRole": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "elasticmapreduce.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+                ],
+            },
+        },
+        "emrEc2Role": {
+            "Type": "AWS::IAM::Role",
+            "Properties": {
+                "AssumeRolePolicyDocument": {
+                    "Version": "2008-10-17",
+                    "Statement": [
+                        {
+                            "Sid": "",
+                            "Effect": "Allow",
+                            "Principal": {"Service": "ec2.amazonaws.com"},
+                            "Action": "sts:AssumeRole",
+                        }
+                    ],
+                },
+                "Path": "/",
+                "ManagedPolicyArns": [
+                    "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role"
+                ],
+            },
+        },
+        "emrEc2InstanceProfile": {
+            "Type": "AWS::IAM::InstanceProfile",
+            "Properties": {"Path": "/", "Roles": [{"Ref": "emrEc2Role"}]},
+        },
+    },
+    "Outputs": {"keyArn": {"Value": {"Fn::GetAtt": ["key", "Arn"]}}},
+}
+
+
+@mock_aws
+def test_create_cluster_with_kerberos_attrs():
+    ec2 = boto3.resource("ec2", region_name="us-east-1")
+    vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
+    subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
+
+    params = [
+        {"ParameterKey": "CrossRealmTrustPrincipalPassword", "ParameterValue": "p2ss"},
+        {"ParameterKey": "KdcAdminPassword", "ParameterValue": "adminp2ss"},
+        {"ParameterKey": "InstanceType", "ParameterValue": "i4s"},
+        {"ParameterKey": "ReleaseLabel", "ParameterValue": "latest"},
+        {"ParameterKey": "EbsRootVolumeSize", "ParameterValue": "15"},
+        {"ParameterKey": "SubnetId", "ParameterValue": subnet.id},
+        {
+            "ParameterKey": "ElasticMapReducePrincipal",
+            "ParameterValue": "emr_principal",
+        },
+        {"ParameterKey": "Ec2Principal", "ParameterValue": "ec2_principal"},
+    ]
+
+    region = "us-east-1"
+    cf = boto3.client("cloudformation", region_name=region)
+    emr = boto3.client("emr", region_name=region)
+    cf.create_stack(
+        StackName="teststack",
+        TemplateBody=json.dumps(template_with_kerberos_attrs),
+        Parameters=params,
+    )
+
+    # Verify resources
+    res = cf.describe_stack_resources(StackName="teststack")["StackResources"]
+    cluster = [r for r in res if r["ResourceType"] == "AWS::EMR::Cluster"][0]
+    cluster_id = cluster["PhysicalResourceId"]
+
+    instance_profile = [
+        r for r in res if r["ResourceType"] == "AWS::IAM::InstanceProfile"
+    ][0]
+    instance_profile_id = instance_profile["PhysicalResourceId"]
+
+    role = [r for r in res if r["ResourceType"] == "AWS::IAM::Role"][0]
+    role_id = role["PhysicalResourceId"]
+
+    # Verify EMR Cluster
+    cl = emr.describe_cluster(ClusterId=cluster_id)["Cluster"]
+    assert cl["Name"] == "CFNtest"
+    assert cl["Ec2InstanceAttributes"]["Ec2SubnetId"] == subnet.id
+    assert cl["Ec2InstanceAttributes"]["IamInstanceProfile"] == instance_profile_id
+    assert cl["ServiceRole"] == role_id
+
+    kerberos = cl["KerberosAttributes"]
+    assert kerberos == {
+        "Realm": "EC2.INTERNAL",
+        "KdcAdminPassword": "adminp2ss",
+        "CrossRealmTrustPrincipalPassword": "p2ss",
+    }