From b8e00554a16aebe4188f2e36eb648da9e36fa72d Mon Sep 17 00:00:00 2001
From: Dirk Avery <dirk.avery@gmail.com>
Date: Tue, 5 Oct 2021 17:10:11 -0400
Subject: [PATCH] tests/emr_cluster: Clean-up, standardize tests

---
 aws/resource_aws_emr_cluster_test.go | 2336 +++++++++++++-------------
 1 file changed, 1183 insertions(+), 1153 deletions(-)

diff --git a/aws/resource_aws_emr_cluster_test.go b/aws/resource_aws_emr_cluster_test.go
index eb8b754c14c..afebd0bec5a 100644
--- a/aws/resource_aws_emr_cluster_test.go
+++ b/aws/resource_aws_emr_cluster_test.go
@@ -80,7 +80,7 @@ func testSweepEmrClusters(region string) error {
 func TestAccAWSEMRCluster_basic(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -123,7 +123,7 @@ func TestAccAWSEMRCluster_additionalInfo(t *testing.T) {
   }
 }`
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -158,7 +158,7 @@ func TestAccAWSEMRCluster_additionalInfo(t *testing.T) {
 func TestAccAWSEMRCluster_disappears(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -181,7 +181,7 @@ func TestAccAWSEMRCluster_disappears(t *testing.T) {
 func TestAccAWSEMRCluster_configurationsJson(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -510,7 +510,7 @@ func TestAccAWSEMRCluster_Ec2Attributes_DefaultManagedSecurityGroups(t *testing.
 	var vpc ec2.Vpc
 
 	rName := acctest.RandomWithPrefix("tf-acc-test")
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	vpcResourceName := "aws_vpc.test"
 
 	resource.ParallelTest(t, resource.TestCase{
@@ -551,7 +551,7 @@ func TestAccAWSEMRCluster_Ec2Attributes_DefaultManagedSecurityGroups(t *testing.
 func TestAccAWSEMRCluster_Kerberos_ClusterDedicatedKdc(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	password := fmt.Sprintf("NeverKeepPasswordsInPlainText%s!", rName)
 	resource.ParallelTest(t, resource.TestCase{
@@ -755,7 +755,7 @@ func TestAccAWSEMRCluster_MasterInstanceGroup_Name(t *testing.T) {
 func TestAccAWSEMRCluster_security_config(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -767,7 +767,7 @@ func TestAccAWSEMRCluster_security_config(t *testing.T) {
 				Config: testAccAWSEmrClusterConfig_SecurityConfiguration(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckAWSEmrClusterExists(resourceName, &cluster),
-					resource.TestCheckResourceAttrPair(resourceName, "security_configuration", "aws_emr_security_configuration.foo", "name"),
+					resource.TestCheckResourceAttrPair(resourceName, "security_configuration", "aws_emr_security_configuration.test", "name"),
 				),
 			},
 			{
@@ -787,7 +787,7 @@ func TestAccAWSEMRCluster_security_config(t *testing.T) {
 func TestAccAWSEMRCluster_Step_Basic(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -825,7 +825,7 @@ func TestAccAWSEMRCluster_Step_Basic(t *testing.T) {
 func TestAccAWSEMRCluster_Step_ConfigMode(t *testing.T) {
 	var cluster1, cluster2, cluster3 emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -891,7 +891,7 @@ func TestAccAWSEMRCluster_Step_ConfigMode(t *testing.T) {
 func TestAccAWSEMRCluster_Step_Multiple(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1033,7 +1033,7 @@ func TestAccAWSEMRCluster_bootstrap_ordering(t *testing.T) {
 func TestAccAWSEMRCluster_terminationProtected(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1090,7 +1090,7 @@ func TestAccAWSEMRCluster_terminationProtected(t *testing.T) {
 func TestAccAWSEMRCluster_keepJob(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1099,7 +1099,7 @@ func TestAccAWSEMRCluster_keepJob(t *testing.T) {
 		CheckDestroy: testAccCheckAWSEmrDestroy,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccAWSEmrClusterConfig_keepJob(rName, "false"),
+				Config: testAccAWSEmrClusterConfig_keepJob(rName, false),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckAWSEmrClusterExists(resourceName, &cluster),
 					resource.TestCheckResourceAttr(resourceName, "keep_job_flow_alive_when_no_steps", "false"),
@@ -1122,7 +1122,7 @@ func TestAccAWSEMRCluster_keepJob(t *testing.T) {
 func TestAccAWSEMRCluster_visibleToAllUsers(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1171,7 +1171,7 @@ func TestAccAWSEMRCluster_visibleToAllUsers(t *testing.T) {
 func TestAccAWSEMRCluster_s3Logging(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	bucketName := fmt.Sprintf("s3n://%s/", rName)
 	resource.ParallelTest(t, resource.TestCase{
@@ -1204,7 +1204,7 @@ func TestAccAWSEMRCluster_s3Logging(t *testing.T) {
 func TestAccAWSEMRCluster_s3LogEncryption(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	bucketName := fmt.Sprintf("s3n://%s/", rName)
 	resource.ParallelTest(t, resource.TestCase{
@@ -1238,7 +1238,7 @@ func TestAccAWSEMRCluster_s3LogEncryption(t *testing.T) {
 func TestAccAWSEMRCluster_tags(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1283,7 +1283,7 @@ func TestAccAWSEMRCluster_tags(t *testing.T) {
 func TestAccAWSEMRCluster_root_volume_size(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1322,7 +1322,7 @@ func TestAccAWSEMRCluster_root_volume_size(t *testing.T) {
 func TestAccAWSEMRCluster_step_concurrency_level(t *testing.T) {
 	var cluster emr.Cluster
 	rName := acctest.RandomWithPrefix("tf-acc-test")
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		ErrorCheck:   testAccErrorCheck(t, emr.EndpointsID),
@@ -1360,7 +1360,7 @@ func TestAccAWSEMRCluster_step_concurrency_level(t *testing.T) {
 func TestAccAWSEMRCluster_ebs_config(t *testing.T) {
 	var cluster emr.Cluster
 	rName := acctest.RandomWithPrefix("tf-acc-test")
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
 		ErrorCheck:   testAccErrorCheck(t, emr.EndpointsID),
@@ -1392,7 +1392,7 @@ func TestAccAWSEMRCluster_ebs_config(t *testing.T) {
 func TestAccAWSEMRCluster_custom_ami_id(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1424,7 +1424,7 @@ func TestAccAWSEMRCluster_custom_ami_id(t *testing.T) {
 func TestAccAWSEMRCluster_InstanceFleet_basic(t *testing.T) {
 	var cluster1, cluster2 emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	subnetResourceName := "aws_subnet.test"
 	subnet2ResourceName := "aws_subnet.test2"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
@@ -1491,7 +1491,7 @@ func TestAccAWSEMRCluster_InstanceFleet_basic(t *testing.T) {
 func TestAccAWSEMRCluster_InstanceFleet_master_only(t *testing.T) {
 	var cluster emr.Cluster
 
-	resourceName := "aws_emr_cluster.tf-test-cluster"
+	resourceName := "aws_emr_cluster.test"
 	rName := acctest.RandomWithPrefix("tf-acc-test")
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -1731,435 +1731,250 @@ func testAccEmrDeleteManagedSecurityGroup(conn *ec2.EC2, securityGroup *ec2.Secu
 	return err
 }
 
-func testAccAWSEmrComposeConfig(mapPublicIPOnLaunch bool, config ...string) string {
-	return composeConfig(append(config, testAccAWSEmrClusterConfigBaseVpc(mapPublicIPOnLaunch))...)
-}
-
-func testAccAWSEmrClusterConfigCurrentPartition() string {
-	return `
-data "aws_partition" "current" {}
-`
-}
-
-func testAccAWSEmrClusterConfig_bootstrap(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigBootstrapActionBucket(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "test" {
-  name          = "%[1]s"
-  release_label = "emr-5.0.0"
-  applications  = ["Hadoop", "Hive"]
-  log_uri       = "s3n://terraform/testlog/"
-
-  master_instance_group {
-    instance_type = "c4.large"
-  }
-
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
-
-  service_role = aws_iam_role.emr_service.arn
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-  ]
-
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
-  }
+// Sub-configs (used by other configs)
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif"
-    args = ["instance.isMaster=true", "echo running on master node"]
+func testAccAWSEmrClusterConfigBaseVpc(rName string, mapPublicIPOnLaunch bool) string {
+	return fmt.Sprintf(`
+data "aws_availability_zones" "available" {
+  # Many instance types are not available in this availability zone
+  exclude_zone_ids = ["usw2-az4"]
+  state            = "available"
+  filter {
+    name   = "opt-in-status"
+    values = ["opt-in-not-required"]
   }
+}
 
-  bootstrap_action {
-    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
-    name = "test"
+resource "aws_vpc" "test" {
+  cidr_block           = "10.0.0.0/16"
+  enable_dns_hostnames = true
 
-    args = ["1",
-      "2",
-      "3",
-      "4",
-      "5",
-      "6",
-      "7",
-      "8",
-      "9",
-      "10",
-    ]
+  tags = {
+    Name = %[1]q
   }
 }
-`, r),
-	)
-}
 
-func testAccAWSEmrClusterConfig_bootstrapAdd(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigBootstrapActionBucket(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "test" {
-  name          = "%[1]s"
-  release_label = "emr-5.0.0"
-  applications  = ["Hadoop", "Hive"]
-  log_uri       = "s3n://terraform/testlog/"
-
-  master_instance_group {
-    instance_type = "c4.large"
-  }
+resource "aws_internet_gateway" "test" {
+  vpc_id = aws_vpc.test.id
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
+  tags = {
+    Name = %[1]q
   }
+}
 
-  service_role = aws_iam_role.emr_service.arn
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-  ]
+resource "aws_security_group" "test" {
+  vpc_id = aws_vpc.test.id
 
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  ingress {
+    from_port = 0
+    protocol  = "-1"
+    self      = true
+    to_port   = 0
   }
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif"
-    args = ["instance.isMaster=true", "echo running on master node"]
+  egress {
+    cidr_blocks = ["0.0.0.0/0"]
+    from_port   = 0
+    protocol    = "-1"
+    to_port     = 0
   }
 
-  bootstrap_action {
-    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
-    name = "test"
-
-    args = ["1",
-      "2",
-      "3",
-      "4",
-      "5",
-      "6",
-      "7",
-      "8",
-      "9",
-      "10",
-    ]
+  tags = {
+    Name = %[1]q
   }
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif-2"
-    args = ["instance.isMaster=true", "echo also running on master node"]
+  # EMR will modify ingress rules
+  lifecycle {
+    ignore_changes = [ingress]
   }
 }
-`, r),
-	)
-}
-
-func testAccAWSEmrClusterConfig_bootstrapReorder(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigBootstrapActionBucket(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "test" {
-  name          = "%[1]s"
-  release_label = "emr-5.0.0"
-  applications  = ["Hadoop", "Hive"]
-  log_uri       = "s3n://terraform/testlog/"
 
-  master_instance_group {
-    instance_type = "c4.large"
-  }
+resource "aws_subnet" "test" {
+  availability_zone       = data.aws_availability_zones.available.names[0]
+  cidr_block              = cidrsubnet(aws_vpc.test.cidr_block, 8, 0)
+  map_public_ip_on_launch = %[2]t
+  vpc_id                  = aws_vpc.test.id
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
+  tags = {
+    Name = %[1]q
   }
+}
 
-  service_role = aws_iam_role.emr_service.arn
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-  ]
+resource "aws_route_table" "test" {
+  vpc_id = aws_vpc.test.id
 
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  route {
+    cidr_block = "0.0.0.0/0"
+    gateway_id = aws_internet_gateway.test.id
   }
+}
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif"
-    args = ["instance.isMaster=true", "echo running on master node"]
-  }
+resource "aws_route_table_association" "test" {
+  route_table_id = aws_route_table.test.id
+  subnet_id      = aws_subnet.test.id
+}
+`, rName, mapPublicIPOnLaunch)
+}
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif-2"
-    args = ["instance.isMaster=true", "echo also running on master node"]
-  }
+func testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_iam_instance_profile" "emr_instance_profile" {
+  name = "%[1]s_profile"
+  role = aws_iam_role.emr_instance_profile.name
+}
 
-  bootstrap_action {
-    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
-    name = "test"
+resource "aws_iam_role" "emr_instance_profile" {
+  name = "%[1]s_profile_role"
 
-    args = ["1",
-      "2",
-      "3",
-      "4",
-      "5",
-      "6",
-      "7",
-      "8",
-      "9",
-      "10",
-    ]
-  }
+  assume_role_policy = <<EOT
+{
+  "Version": "2008-10-17",
+  "Statement": [
+    {
+      "Sid": "",
+      "Effect": "Allow",
+      "Principal": {
+        "Service": "ec2.${data.aws_partition.current.dns_suffix}"
+      },
+      "Action": "sts:AssumeRole"
+    }
+  ]
 }
-`, r),
-	)
+EOT
 }
 
-func testAccAWSEmrClusterConfig(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-4.6.0"
-  applications  = ["Spark"]
+resource "aws_iam_role_policy_attachment" "emr_instance_profile" {
+  role       = aws_iam_role.emr_instance_profile.id
+  policy_arn = aws_iam_policy.emr_instance_profile.arn
+}
 
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
-  }
+resource "aws_iam_policy" "emr_instance_profile" {
+  name = "%[1]s_profile"
 
-  master_instance_group {
-    instance_type = "c4.large"
-  }
-
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
-
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
-  }
-
-  keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
-
-  scale_down_behavior = "TERMINATE_AT_TASK_COMPLETION"
+  policy = <<EOT
+{
+    "Version": "2012-10-17",
+    "Statement": [{
+        "Effect": "Allow",
+        "Resource": "*",
+        "Action": [
+            "cloudwatch:*",
+            "dynamodb:*",
+            "ec2:Describe*",
+            "elasticmapreduce:Describe*",
+            "elasticmapreduce:ListBootstrapActions",
+            "elasticmapreduce:ListClusters",
+            "elasticmapreduce:ListInstanceGroups",
+            "elasticmapreduce:ListInstances",
+            "elasticmapreduce:ListSteps",
+            "kinesis:CreateStream",
+            "kinesis:DeleteStream",
+            "kinesis:DescribeStream",
+            "kinesis:GetRecords",
+            "kinesis:GetShardIterator",
+            "kinesis:MergeShards",
+            "kinesis:PutRecord",
+            "kinesis:SplitShard",
+            "rds:Describe*",
+            "s3:*",
+            "sdb:*",
+            "sns:*",
+            "sqs:*"
+        ]
+    }]
+}
+EOT
+}
+`, rName)
+}
 
-  configurations = "test-fixtures/emr_configurations.json"
+func testAccAWSEmrClusterConfigIAMServiceRoleBase(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_iam_role" "emr_service" {
+  name = "%[1]s_default_role"
 
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-    aws_iam_role_policy_attachment.emr_autoscaling_role,
+  assume_role_policy = <<EOT
+{
+  "Version": "2008-10-17",
+  "Statement": [
+    {
+      "Sid": "",
+      "Effect": "Allow",
+      "Principal": {
+        "Service": "elasticmapreduce.${data.aws_partition.current.dns_suffix}"
+      },
+      "Action": "sts:AssumeRole"
+    }
   ]
-
-  service_role         = aws_iam_role.emr_service.arn
-  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
-  ebs_root_volume_size = 21
 }
-`, r),
-	)
+EOT
 }
 
-func testAccAWSEmrClusterConfigAdditionalInfo(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-4.6.0"
-  applications  = ["Spark"]
-
-  additional_info = <<EOF
-{
-	"instanceAwsClientConfiguration": {
-		"proxyPort": 8099,
-		"proxyHost": "myproxy.example.com"
-	}
+resource "aws_iam_role_policy_attachment" "emr_service" {
+  role       = aws_iam_role.emr_service.id
+  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
 }
-EOF
 
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
-  }
+`, rName)
+}
 
-  master_instance_group {
-    instance_type = "c4.large"
-  }
+func testAccAWSEmrClusterConfigIAMAutoscalingRole(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_iam_role" "emr_autoscaling_role" {
+  name               = "%[1]s_autoscaling_role"
+  assume_role_policy = data.aws_iam_policy_document.emr_autoscaling_role.json
+}
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
+data "aws_iam_policy_document" "emr_autoscaling_role" {
+  statement {
+    effect  = "Allow"
+    actions = ["sts:AssumeRole"]
 
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
+    principals {
+      type        = "Service"
+      identifiers = ["elasticmapreduce.${data.aws_partition.current.dns_suffix}", "application-autoscaling.${data.aws_partition.current.dns_suffix}"]
+    }
   }
+}
 
-  keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
-
-  scale_down_behavior = "TERMINATE_AT_TASK_COMPLETION"
-
-  configurations = "test-fixtures/emr_configurations.json"
-
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-    aws_iam_role_policy_attachment.emr_autoscaling_role,
-  ]
-
-  service_role         = aws_iam_role.emr_service.arn
-  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
-  ebs_root_volume_size = 21
+resource "aws_iam_role_policy_attachment" "emr_autoscaling_role" {
+  role       = aws_iam_role.emr_autoscaling_role.name
+  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceforAutoScalingRole"
 }
-`, r),
-	)
+`, rName)
 }
 
-func testAccAWSEmrClusterConfigConfigurationsJson(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-4.6.0"
-  applications  = ["Hadoop", "Spark"]
-
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
-  }
-
-  master_instance_group {
-    instance_type = "c4.large"
-  }
-
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
-
-  keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
+func testAccAWSEmrClusterConfigBootstrapActionBucket(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_s3_bucket" "tester" {
+  bucket = %[1]q
+  acl    = "public-read"
+}
 
-  configurations_json = <<EOF
-[
-   {
-     "Classification": "hadoop-env",
-     "Configurations": [
-       {
-         "Classification": "export",
-         "Properties": {
-           "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"
-         }
-       }
-     ],
-     "Properties": {}
-   },
-   {
-     "Classification": "spark-env",
-     "Configurations": [
-       {
-         "Classification": "export",
-         "Properties": {
-           "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"
-         }
-       }
-     ],
-     "Properties": {}
-   }
-]
+resource "aws_s3_bucket_object" "testobject" {
+  bucket  = aws_s3_bucket.tester.bucket
+  key     = "testscript.sh"
+  content = <<EOF
+#!/bin/bash
+echo $@
 EOF
 
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-  ]
-
-  service_role         = aws_iam_role.emr_service.arn
-  ebs_root_volume_size = 21
-}
-`, r),
-	)
-}
 
-func testAccAWSEmrClusterConfig_Kerberos_ClusterDedicatedKdc(r string, password string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
-resource "aws_emr_security_configuration" "foo" {
-  configuration = <<EOF
-{
-  "AuthenticationConfiguration": {
-    "KerberosConfiguration": {
-      "Provider": "ClusterDedicatedKdc",
-      "ClusterDedicatedKdcConfiguration": {
-        "TicketLifetimeInHours": 24
-      }
-    }
-  }
+  acl = "public-read"
 }
-EOF
+`, rName)
 }
 
-resource "aws_emr_cluster" "tf-test-cluster" {
+func testAccAWSEmrClusterConfig_Step(rName string, stepConfig string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
+resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
-  name                              = "%[1]s"
+  log_uri                           = "s3://${aws_s3_bucket.test.bucket}/"
+  name                              = %[1]q
   release_label                     = "emr-5.12.0"
-  security_configuration            = aws_emr_security_configuration.foo.name
   service_role                      = "EMR_DefaultRole"
   termination_protection            = false
 
@@ -2179,112 +1994,115 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     subnet_id                         = aws_subnet.test.id
   }
 
-  kerberos_attributes {
-    kdc_admin_password = "%[2]s"
-    realm              = "EC2.INTERNAL"
-  }
+  %[2]s
 
   depends_on = [aws_route_table_association.test]
 }
-`, r, password))
-}
 
-func testAccAWSEmrClusterConfig_SecurityConfiguration(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-5.5.0"
-  applications  = ["Spark"]
+resource "aws_s3_bucket" "test" {
+  bucket        = %[1]q
+  force_destroy = true
+}
+`, rName, stepConfig))
+}
 
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
-  }
+func testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_iam_role" "emr_service" {
+  name = "%[1]s_default_role"
 
-  master_instance_group {
-    instance_type = "c4.large"
-  }
-
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
-
-  security_configuration = aws_emr_security_configuration.foo.name
-
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
-  }
-
-  keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
-
-  configurations = "test-fixtures/emr_configurations.json"
-
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-    aws_iam_role_policy_attachment.emr_autoscaling_role,
-  ]
-
-  service_role     = aws_iam_role.emr_service.arn
-  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
-}
-
-resource "aws_emr_security_configuration" "foo" {
-  configuration = <<EOF
-{
-  "EncryptionConfiguration": {
-    "AtRestEncryptionConfiguration": {
-      "S3EncryptionConfiguration": {
-        "EncryptionMode": "SSE-S3"
-      },
-      "LocalDiskEncryptionConfiguration": {
-        "EncryptionKeyProviderType": "AwsKms",
-        "AwsKmsKey": "${aws_kms_key.foo.arn}"
-      }
-    },
-    "EnableInTransitEncryption": false,
-    "EnableAtRestEncryption": true
-  }
-}
-EOF
-}
-
-resource "aws_kms_key" "foo" {
-  description             = "Terraform %[1]s"
-  deletion_window_in_days = 7
-
-  policy = <<POLICY
+  assume_role_policy = <<EOT
 {
-  "Version": "2012-10-17",
-  "Id": "kms-tf-1",
+  "Version": "2008-10-17",
   "Statement": [
     {
-      "Sid": "Enable IAM User Permissions",
+      "Sid": "",
       "Effect": "Allow",
       "Principal": {
-        "AWS": "*"
+        "Service": "elasticmapreduce.${data.aws_partition.current.dns_suffix}"
       },
-      "Action": "kms:*",
-      "Resource": "*"
+      "Action": "sts:AssumeRole"
     }
   ]
 }
-POLICY
+EOT
+}
+
+resource "aws_iam_role_policy_attachment" "emr_service" {
+  role       = aws_iam_role.emr_service.id
+  policy_arn = aws_iam_policy.emr_service.arn
+}
+
+resource "aws_iam_policy" "emr_service" {
+  name = "%[1]s_emr"
+
+  policy = <<EOT
+{
+    "Version": "2012-10-17",
+    "Statement": [{
+        "Effect": "Allow",
+        "Resource": "*",
+        "Action": [
+            "ec2:AuthorizeSecurityGroupEgress",
+            "ec2:AuthorizeSecurityGroupIngress",
+            "ec2:CancelSpotInstanceRequests",
+            "ec2:CreateNetworkInterface",
+            "ec2:CreateSecurityGroup",
+            "ec2:CreateTags",
+            "ec2:DeleteNetworkInterface",
+            "ec2:DeleteSecurityGroup",
+            "ec2:DeleteTags",
+            "ec2:DescribeAvailabilityZones",
+            "ec2:DescribeAccountAttributes",
+            "ec2:DescribeDhcpOptions",
+            "ec2:DescribeImages",
+            "ec2:DescribeInstanceStatus",
+            "ec2:DescribeInstances",
+            "ec2:DescribeKeyPairs",
+            "ec2:DescribeNetworkAcls",
+            "ec2:DescribeNetworkInterfaces",
+            "ec2:DescribePrefixLists",
+            "ec2:DescribeRouteTables",
+            "ec2:DescribeSecurityGroups",
+            "ec2:DescribeSpotInstanceRequests",
+            "ec2:DescribeSpotPriceHistory",
+            "ec2:DescribeSubnets",
+            "ec2:DescribeVpcAttribute",
+            "ec2:DescribeVpcEndpoints",
+            "ec2:DescribeVpcEndpointServices",
+            "ec2:DescribeVpcs",
+            "ec2:DetachNetworkInterface",
+            "ec2:ModifyImageAttribute",
+            "ec2:ModifyInstanceAttribute",
+            "ec2:RequestSpotInstances",
+            "ec2:RevokeSecurityGroupEgress",
+            "ec2:RunInstances",
+            "ec2:TerminateInstances",
+            "ec2:DeleteVolume",
+            "ec2:DescribeVolumeStatus",
+            "ec2:DescribeVolumes",
+            "ec2:DetachVolume",
+            "iam:GetRole",
+            "iam:GetRolePolicy",
+            "iam:ListInstanceProfiles",
+            "iam:ListRolePolicies",
+            "iam:PassRole",
+            "s3:CreateBucket",
+            "s3:Get*",
+            "s3:List*",
+            "sdb:BatchPutAttributes",
+            "sdb:Select",
+            "sqs:CreateQueue",
+            "sqs:Delete*",
+            "sqs:GetQueue*",
+            "sqs:PurgeQueue",
+            "sqs:ReceiveMessage"
+        ]
+    }]
 }
-`, r))
+EOT
+}
+`, rName)
 }
 
 const testAccAWSEmrClusterConfig_Step_DebugLoggingStep = `
@@ -2313,33 +2131,28 @@ const testAccAWSEmrClusterConfig_Step_SparkStep = `
   }
 `
 
-func testAccAWSEmrClusterConfig_Step_Multiple(r string) string {
-	stepConfig := testAccAWSEmrClusterConfig_Step_DebugLoggingStep + testAccAWSEmrClusterConfig_Step_SparkStep
-	return testAccAWSEmrClusterConfig_Step(r, stepConfig)
-}
-
-func testAccAWSEmrClusterConfig_Step_NoBlocks(r string) string {
-	return testAccAWSEmrClusterConfig_Step(r, "")
-}
+// Configs
 
-func testAccAWSEmrClusterConfig_Step_Single(r string) string {
-	return testAccAWSEmrClusterConfig_Step(r, testAccAWSEmrClusterConfig_Step_DebugLoggingStep)
-}
+func testAccAWSEmrClusterConfig(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
 
-func testAccAWSEmrClusterConfig_Step_Zeroed(r string) string {
-	return testAccAWSEmrClusterConfig_Step(r, "step = []")
-}
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-4.6.0"
+  applications  = ["Spark"]
 
-func testAccAWSEmrClusterConfig_Step(r string, stepConfig string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  applications                      = ["Spark"]
-  keep_job_flow_alive_when_no_steps = true
-  log_uri                           = "s3://${aws_s3_bucket.test.bucket}/"
-  name                              = "%[1]s"
-  release_label                     = "emr-5.12.0"
-  service_role                      = "EMR_DefaultRole"
-  termination_protection            = false
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  }
 
   master_instance_group {
     instance_type = "c4.large"
@@ -2350,93 +2163,245 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
-  ec2_attributes {
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = "EMR_EC2_DefaultRole"
-    subnet_id                         = aws_subnet.test.id
+  tags = {
+    role     = "rolename"
+    dns_zone = "env_zone"
+    env      = "env"
+    name     = "name-env"
   }
 
-  %[2]s
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
 
-  depends_on = [aws_route_table_association.test]
-}
+  scale_down_behavior = "TERMINATE_AT_TASK_COMPLETION"
 
-resource "aws_s3_bucket" "test" {
-  bucket        = "%[1]s"
-  force_destroy = true
-}
-`, r, stepConfig))
-}
+  configurations = "test-fixtures/emr_configurations.json"
 
-func testAccAWSEmrClusterConfigCoreInstanceGroupAutoscalingPolicy(rName, autoscalingPolicy string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		fmt.Sprintf(`
-data "aws_iam_policy_document" "test" {
-  statement {
-    actions = ["sts:AssumeRole"]
-    effect  = "Allow"
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+    aws_iam_role_policy_attachment.emr_autoscaling_role,
+  ]
 
-    principals {
-      identifiers = [
-        "application-autoscaling.${data.aws_partition.current.dns_suffix}",
-        "elasticmapreduce.${data.aws_partition.current.dns_suffix}",
-      ]
-      type = "Service"
-    }
-  }
+  service_role         = aws_iam_role.emr_service.arn
+  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
+  ebs_root_volume_size = 21
 }
-
-resource "aws_iam_role" "test" {
-  name               = %[1]q
-  assume_role_policy = data.aws_iam_policy_document.test.json
+`, rName))
 }
 
-resource "aws_iam_role_policy_attachment" "test" {
-  role       = aws_iam_role.test.name
-  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceforAutoScalingRole"
-}
+func testAccAWSEmrClusterConfigAdditionalInfo(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
 
 resource "aws_emr_cluster" "test" {
-  applications                      = ["Spark"]
-  autoscaling_role                  = aws_iam_role.test.arn
-  keep_job_flow_alive_when_no_steps = true
-  name                              = %[1]q
-  release_label                     = "emr-5.12.0"
-  service_role                      = "EMR_DefaultRole"
+  name          = %[1]q
+  release_label = "emr-4.6.0"
+  applications  = ["Spark"]
+
+  additional_info = <<EOF
+{
+	"instanceAwsClientConfiguration": {
+		"proxyPort": 8099,
+		"proxyHost": "myproxy.example.com"
+	}
+}
+EOF
 
   ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
     emr_managed_master_security_group = aws_security_group.test.id
     emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = "EMR_EC2_DefaultRole"
-    subnet_id                         = aws_subnet.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
   }
 
   master_instance_group {
-    instance_type = "m4.large"
+    instance_type = "c4.large"
   }
 
   core_instance_group {
-    autoscaling_policy = <<POLICY
-%[2]s
-POLICY
-    instance_type      = "m4.large"
+    instance_count = 1
+    instance_type  = "c4.large"
   }
 
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.test,
-  ]
-}
-`, rName, autoscalingPolicy),
-	)
-}
-
-func testAccAWSEmrClusterConfigCoreInstanceGroupAutoscalingPolicyRemoved(rName string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
+  tags = {
+    role     = "rolename"
+    dns_zone = "env_zone"
+    env      = "env"
+    name     = "name-env"
+  }
+
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
+
+  scale_down_behavior = "TERMINATE_AT_TASK_COMPLETION"
+
+  configurations = "test-fixtures/emr_configurations.json"
+
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+    aws_iam_role_policy_attachment.emr_autoscaling_role,
+  ]
+
+  service_role         = aws_iam_role.emr_service.arn
+  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
+  ebs_root_volume_size = 21
+}
+`, rName))
+}
+
+func testAccAWSEmrClusterConfigConfigurationsJson(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-4.6.0"
+  applications  = ["Hadoop", "Spark"]
+
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  }
+
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
+
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
+
+  configurations_json = <<EOF
+[
+   {
+     "Classification": "hadoop-env",
+     "Configurations": [
+       {
+         "Classification": "export",
+         "Properties": {
+           "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"
+         }
+       }
+     ],
+     "Properties": {}
+   },
+   {
+     "Classification": "spark-env",
+     "Configurations": [
+       {
+         "Classification": "export",
+         "Properties": {
+           "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"
+         }
+       }
+     ],
+     "Properties": {}
+   }
+]
+EOF
+
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+  ]
+
+  service_role         = aws_iam_role.emr_service.arn
+  ebs_root_volume_size = 21
+}
+`, rName))
+}
+
+func testAccAWSEmrClusterConfigCoreInstanceGroupAutoscalingPolicy(rName, autoscalingPolicy string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+data "aws_iam_policy_document" "test" {
+  statement {
+    actions = ["sts:AssumeRole"]
+    effect  = "Allow"
+
+    principals {
+      identifiers = [
+        "application-autoscaling.${data.aws_partition.current.dns_suffix}",
+        "elasticmapreduce.${data.aws_partition.current.dns_suffix}",
+      ]
+      type = "Service"
+    }
+  }
+}
+
+resource "aws_iam_role" "test" {
+  name               = %[1]q
+  assume_role_policy = data.aws_iam_policy_document.test.json
+}
+
+resource "aws_iam_role_policy_attachment" "test" {
+  role       = aws_iam_role.test.name
+  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceforAutoScalingRole"
+}
+
+resource "aws_emr_cluster" "test" {
+  applications                      = ["Spark"]
+  autoscaling_role                  = aws_iam_role.test.arn
+  keep_job_flow_alive_when_no_steps = true
+  name                              = %[1]q
+  release_label                     = "emr-5.12.0"
+  service_role                      = "EMR_DefaultRole"
+
+  ec2_attributes {
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = "EMR_EC2_DefaultRole"
+    subnet_id                         = aws_subnet.test.id
+  }
+
+  master_instance_group {
+    instance_type = "m4.large"
+  }
+
+  core_instance_group {
+    autoscaling_policy = <<POLICY
+%[2]s
+POLICY
+    instance_type      = "m4.large"
+  }
+
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.test,
+  ]
+}
+`, rName, autoscalingPolicy))
+}
+
+func testAccAWSEmrClusterConfigCoreInstanceGroupAutoscalingPolicyRemoved(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
 		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
 data "aws_iam_policy_document" "test" {
   statement {
     actions = ["sts:AssumeRole"]
@@ -2490,12 +2455,13 @@ resource "aws_emr_cluster" "test" {
     aws_iam_role_policy_attachment.test,
   ]
 }
-`, rName),
-	)
+`, rName))
 }
 
 func testAccAWSEmrClusterConfigCoreInstanceGroupBidPrice(rName, bidPrice string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2521,12 +2487,13 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, bidPrice),
-	)
+`, rName, bidPrice))
 }
 
 func testAccAWSEmrClusterConfigCoreInstanceGroupInstanceCount(rName string, instanceCount int) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2552,12 +2519,13 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, instanceCount),
-	)
+`, rName, instanceCount))
 }
 
 func testAccAWSEmrClusterConfigCoreInstanceGroupInstanceType(rName, instanceType string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2582,12 +2550,13 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, instanceType),
-	)
+`, rName, instanceType))
 }
 
 func testAccAWSEmrClusterConfigCoreInstanceGroupName(rName, instanceGroupName string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2613,13 +2582,14 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, instanceGroupName),
-	)
+`, rName, instanceGroupName))
 }
 
 func testAccAWSEmrClusterConfigEc2AttributesDefaultManagedSecurityGroups(rName string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
+resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
   name                              = %[1]q
@@ -2637,12 +2607,67 @@ resource "aws_emr_cluster" "tf-test-cluster" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName),
-	)
+`, rName))
+}
+
+func testAccAWSEmrClusterConfig_Kerberos_ClusterDedicatedKdc(rName string, password string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
+resource "aws_emr_security_configuration" "test" {
+  configuration = <<EOF
+{
+  "AuthenticationConfiguration": {
+    "KerberosConfiguration": {
+      "Provider": "ClusterDedicatedKdc",
+      "ClusterDedicatedKdcConfiguration": {
+        "TicketLifetimeInHours": 24
+      }
+    }
+  }
+}
+EOF
+}
+
+resource "aws_emr_cluster" "test" {
+  applications                      = ["Spark"]
+  keep_job_flow_alive_when_no_steps = true
+  name                              = %[1]q
+  release_label                     = "emr-5.12.0"
+  security_configuration            = aws_emr_security_configuration.test.name
+  service_role                      = "EMR_DefaultRole"
+  termination_protection            = false
+
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
+
+  ec2_attributes {
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = "EMR_EC2_DefaultRole"
+    subnet_id                         = aws_subnet.test.id
+  }
+
+  kerberos_attributes {
+    kdc_admin_password = "%[2]s"
+    realm              = "EC2.INTERNAL"
+  }
+
+  depends_on = [aws_route_table_association.test]
+}
+`, rName, password))
 }
 
 func testAccAWSEmrClusterConfigMasterInstanceGroupBidPrice(rName, bidPrice string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2664,12 +2689,13 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, bidPrice),
-	)
+`, rName, bidPrice))
 }
 
 func testAccAWSEmrClusterConfigMasterInstanceGroupInstanceCount(rName string, instanceCount int) string {
-	return testAccAWSEmrComposeConfig(true, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, true),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2699,12 +2725,13 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, instanceCount),
-	)
+`, rName, instanceCount))
 }
 
 func testAccAWSEmrClusterConfigMasterInstanceGroupInstanceType(rName, instanceType string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2725,12 +2752,13 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, instanceType),
-	)
+`, rName, instanceType))
 }
 
 func testAccAWSEmrClusterConfigMasterInstanceGroupName(rName, instanceGroupName string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
 resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
@@ -2752,20 +2780,21 @@ resource "aws_emr_cluster" "test" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, instanceGroupName),
-	)
+`, rName, instanceGroupName))
 }
 
-func testAccAWSEmrClusterConfigTerminationPolicy(r string, term string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+func testAccAWSEmrClusterConfig_SecurityConfiguration(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-4.6.0"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-5.5.0"
   applications  = ["Spark"]
 
   ec2_attributes {
@@ -2784,6 +2813,8 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
+  security_configuration = aws_emr_security_configuration.test.name
+
   tags = {
     role     = "rolename"
     dns_zone = "env_zone"
@@ -2792,7 +2823,7 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   }
 
   keep_job_flow_alive_when_no_steps = true
-  termination_protection            = %[2]s
+  termination_protection            = false
 
   configurations = "test-fixtures/emr_configurations.json"
 
@@ -2806,83 +2837,284 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   service_role     = aws_iam_role.emr_service.arn
   autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
 }
-`, r, term),
-	)
-}
-
-func testAccAWSEmrClusterConfig_keepJob(r string, keepJob string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-4.6.0"
-  applications  = ["Spark"]
 
-  ec2_attributes {
-    subnet_id                         = aws_subnet.test.id
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+resource "aws_emr_security_configuration" "test" {
+  configuration = <<EOF
+{
+  "EncryptionConfiguration": {
+    "AtRestEncryptionConfiguration": {
+      "S3EncryptionConfiguration": {
+        "EncryptionMode": "SSE-S3"
+      },
+      "LocalDiskEncryptionConfiguration": {
+        "EncryptionKeyProviderType": "AwsKms",
+        "AwsKmsKey": "${aws_kms_key.test.arn}"
+      }
+    },
+    "EnableInTransitEncryption": false,
+    "EnableAtRestEncryption": true
   }
+}
+EOF
+}
 
-  master_instance_group {
-    instance_type = "c4.large"
-  }
+resource "aws_kms_key" "test" {
+  description             = "Terraform %[1]s"
+  deletion_window_in_days = 7
 
-  core_instance_group {
+  policy = <<POLICY
+{
+  "Version": "2012-10-17",
+  "Id": "kms-tf-1",
+  "Statement": [
+    {
+      "Sid": "Enable IAM User Permissions",
+      "Effect": "Allow",
+      "Principal": {
+        "AWS": "*"
+      },
+      "Action": "kms:*",
+      "Resource": "*"
+    }
+  ]
+}
+POLICY
+}
+`, rName))
+}
+
+func testAccAWSEmrClusterConfig_Step_Single(rName string) string {
+	return testAccAWSEmrClusterConfig_Step(rName, testAccAWSEmrClusterConfig_Step_DebugLoggingStep)
+}
+
+func testAccAWSEmrClusterConfig_Step_NoBlocks(rName string) string {
+	return testAccAWSEmrClusterConfig_Step(rName, "")
+}
+
+func testAccAWSEmrClusterConfig_Step_Zeroed(rName string) string {
+	return testAccAWSEmrClusterConfig_Step(rName, "step = []")
+}
+
+func testAccAWSEmrClusterConfig_Step_Multiple(rName string) string {
+	stepConfig := testAccAWSEmrClusterConfig_Step_DebugLoggingStep + testAccAWSEmrClusterConfig_Step_SparkStep
+	return testAccAWSEmrClusterConfig_Step(rName, stepConfig)
+}
+
+func testAccAWSEmrClusterConfig_bootstrap(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigBootstrapActionBucket(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-5.0.0"
+  applications  = ["Hadoop", "Hive"]
+  log_uri       = "s3n://terraform/testlog/"
+
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  core_instance_group {
     instance_count = 1
     instance_type  = "c4.large"
   }
 
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
+  service_role = aws_iam_role.emr_service.arn
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+  ]
+
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
   }
 
-  keep_job_flow_alive_when_no_steps = %s
-  termination_protection            = false
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif"
+    args = ["instance.isMaster=true", "echo running on master node"]
+  }
 
-  step {
-    action_on_failure = "CONTINUE"
-    name              = "Sleep Step"
+  bootstrap_action {
+    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
+    name = "test"
 
-    hadoop_jar_step {
-      jar  = "command-runner.jar"
-      args = ["/bin/sleep", "60"]
-    }
+    args = ["1",
+      "2",
+      "3",
+      "4",
+      "5",
+      "6",
+      "7",
+      "8",
+      "9",
+      "10",
+    ]
   }
+}
+`, rName))
+}
 
-  configurations = "test-fixtures/emr_configurations.json"
+func testAccAWSEmrClusterConfig_bootstrapAdd(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigBootstrapActionBucket(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-5.0.0"
+  applications  = ["Hadoop", "Hive"]
+  log_uri       = "s3n://terraform/testlog/"
 
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
+
+  service_role = aws_iam_role.emr_service.arn
   depends_on = [
     aws_route_table_association.test,
     aws_iam_role_policy_attachment.emr_service,
     aws_iam_role_policy_attachment.emr_instance_profile,
-    aws_iam_role_policy_attachment.emr_autoscaling_role,
   ]
 
-  service_role     = aws_iam_role.emr_service.arn
-  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  }
+
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif"
+    args = ["instance.isMaster=true", "echo running on master node"]
+  }
+
+  bootstrap_action {
+    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
+    name = "test"
+
+    args = ["1",
+      "2",
+      "3",
+      "4",
+      "5",
+      "6",
+      "7",
+      "8",
+      "9",
+      "10",
+    ]
+  }
+
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif-2"
+    args = ["instance.isMaster=true", "echo also running on master node"]
+  }
+}
+`, rName))
+}
+
+func testAccAWSEmrClusterConfig_bootstrapReorder(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigBootstrapActionBucket(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-5.0.0"
+  applications  = ["Hadoop", "Hive"]
+  log_uri       = "s3n://terraform/testlog/"
+
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
+
+  service_role = aws_iam_role.emr_service.arn
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+  ]
+
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  }
+
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif"
+    args = ["instance.isMaster=true", "echo running on master node"]
+  }
+
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif-2"
+    args = ["instance.isMaster=true", "echo also running on master node"]
+  }
+
+  bootstrap_action {
+    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
+    name = "test"
+
+    args = ["1",
+      "2",
+      "3",
+      "4",
+      "5",
+      "6",
+      "7",
+      "8",
+      "9",
+      "10",
+    ]
+  }
 }
-`, r, keepJob),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigVisibleToAllUsersUpdated(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+func testAccAWSEmrClusterConfigTerminationPolicy(rName string, term string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-4.6.0"
   applications  = ["Spark"]
 
@@ -2910,7 +3142,7 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   }
 
   keep_job_flow_alive_when_no_steps = true
-  visible_to_all_users              = false
+  termination_protection            = %[2]s
 
   configurations = "test-fixtures/emr_configurations.json"
 
@@ -2924,19 +3156,20 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   service_role     = aws_iam_role.emr_service.arn
   autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
 }
-`, r),
-	)
+`, rName, term))
 }
 
-func testAccAWSEmrClusterConfigUpdatedTags(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+func testAccAWSEmrClusterConfig_keepJob(rName string, keepJob bool) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-4.6.0"
   applications  = ["Spark"]
 
@@ -2957,14 +3190,25 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   }
 
   tags = {
-    dns_zone = "new_zone"
-    Env      = "production"
+    role     = "rolename"
+    dns_zone = "env_zone"
+    env      = "env"
     name     = "name-env"
   }
 
-  keep_job_flow_alive_when_no_steps = true
+  keep_job_flow_alive_when_no_steps = %[2]t
   termination_protection            = false
 
+  step {
+    action_on_failure = "CONTINUE"
+    name              = "Sleep Step"
+
+    hadoop_jar_step {
+      jar  = "command-runner.jar"
+      args = ["/bin/sleep", "60"]
+    }
+  }
+
   configurations = "test-fixtures/emr_configurations.json"
 
   depends_on = [
@@ -2977,19 +3221,20 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   service_role     = aws_iam_role.emr_service.arn
   autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
 }
-`, r),
-	)
+`, rName, keepJob))
 }
 
-func testAccAWSEmrClusterConfigUpdatedRootVolumeSize(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+func testAccAWSEmrClusterConfigVisibleToAllUsersUpdated(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-4.6.0"
   applications  = ["Spark"]
 
@@ -3017,7 +3262,7 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   }
 
   keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
+  visible_to_all_users              = false
 
   configurations = "test-fixtures/emr_configurations.json"
 
@@ -3028,27 +3273,27 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     aws_iam_role_policy_attachment.emr_autoscaling_role,
   ]
 
-  service_role         = aws_iam_role.emr_service.arn
-  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
-  ebs_root_volume_size = 48
+  service_role     = aws_iam_role.emr_service.arn
+  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
 }
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigS3Logging(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
+func testAccAWSEmrClusterConfigS3Logging(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
 		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
 resource "aws_s3_bucket" "test" {
-  bucket        = "%[1]s"
+  bucket        = %[1]q
   force_destroy = true
 }
 
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-4.6.0"
   applications  = ["Spark"]
 
@@ -3077,22 +3322,23 @@ resource "aws_emr_cluster" "tf-test-cluster" {
 }
 
 data "aws_caller_identity" "current" {}
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigS3Encryption(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
+func testAccAWSEmrClusterConfigS3Encryption(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
 		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
 resource "aws_s3_bucket" "test" {
-  bucket        = "%[1]s"
+  bucket        = %[1]q
   force_destroy = true
 }
 
-resource "aws_kms_key" "foo" {
+resource "aws_kms_key" "test" {
   description = "Terraform acc test %[1]s"
 
   policy = <<POLICY
@@ -3114,8 +3360,8 @@ resource "aws_kms_key" "foo" {
 POLICY
 }
 
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-5.32.0"
   applications  = ["Spark"]
 
@@ -3131,7 +3377,7 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
-  log_encryption_kms_key_id = aws_kms_key.foo.arn
+  log_encryption_kms_key_id = aws_kms_key.test.arn
   log_uri                   = "s3://${aws_s3_bucket.test.bucket}/"
 
   ec2_attributes {
@@ -3145,20 +3391,21 @@ resource "aws_emr_cluster" "tf-test-cluster" {
 }
 
 data "aws_caller_identity" "current" {}
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigCustomAmiID(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+func testAccAWSEmrClusterConfigUpdatedTags(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  release_label = "emr-5.7.0"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-4.6.0"
   applications  = ["Spark"]
 
   ec2_attributes {
@@ -3172,429 +3419,128 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type = "c4.large"
   }
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
-
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
-  }
-
-  keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
-
-  configurations = "test-fixtures/emr_configurations.json"
-
-  depends_on = [
-    aws_route_table_association.test,
-    aws_iam_role_policy_attachment.emr_service,
-    aws_iam_role_policy_attachment.emr_instance_profile,
-    aws_iam_role_policy_attachment.emr_autoscaling_role,
-  ]
-
-  service_role         = aws_iam_role.emr_service.arn
-  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
-  ebs_root_volume_size = 48
-  custom_ami_id        = data.aws_ami.emr-custom-ami.id
-}
-
-data "aws_ami" "emr-custom-ami" {
-  most_recent = true
-  owners      = ["amazon"]
-
-  filter {
-    name   = "name"
-    values = ["amzn-ami-hvm-*"]
-  }
-
-  filter {
-    name   = "architecture"
-    values = ["x86_64"]
-  }
-
-  filter {
-    name   = "root-device-type"
-    values = ["ebs"]
-  }
-
-  filter {
-    name   = "virtualization-type"
-    values = ["hvm"]
-  }
-}
-`, r),
-	)
-}
-
-func testAccAWSEmrClusterConfigStepConcurrencyLevel(rName string, stepConcurrencyLevel int) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  applications                      = ["Spark"]
-  keep_job_flow_alive_when_no_steps = true
-  name                              = %[1]q
-  release_label                     = "emr-5.28.0"
-  service_role                      = "EMR_DefaultRole"
-
-  ec2_attributes {
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    instance_profile                  = "EMR_EC2_DefaultRole"
-    subnet_id                         = aws_subnet.test.id
-  }
-
-  master_instance_group {
-    instance_type = "m4.large"
-  }
-
-  step_concurrency_level = %[2]d
-
-  depends_on = [aws_route_table_association.test]
-}
-`, rName, stepConcurrencyLevel),
-	)
-}
-
-func testAccAWSEmrClusterConfigBaseVpc(mapPublicIPOnLaunch bool) string {
-	return fmt.Sprintf(`
-data "aws_availability_zones" "available" {
-  # Many instance types are not available in this availability zone
-  exclude_zone_ids = ["usw2-az4"]
-  state            = "available"
-  filter {
-    name   = "opt-in-status"
-    values = ["opt-in-not-required"]
-  }
-}
-
-resource "aws_vpc" "test" {
-  cidr_block           = "10.0.0.0/16"
-  enable_dns_hostnames = true
-
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
-  }
-}
-
-resource "aws_internet_gateway" "test" {
-  vpc_id = aws_vpc.test.id
-
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
-  }
-}
-
-resource "aws_security_group" "test" {
-  vpc_id = aws_vpc.test.id
-
-  ingress {
-    from_port = 0
-    protocol  = "-1"
-    self      = true
-    to_port   = 0
-  }
-
-  egress {
-    cidr_blocks = ["0.0.0.0/0"]
-    from_port   = 0
-    protocol    = "-1"
-    to_port     = 0
-  }
-
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
-  }
-
-  # EMR will modify ingress rules
-  lifecycle {
-    ignore_changes = [ingress]
-  }
-}
-
-resource "aws_subnet" "test" {
-  availability_zone       = data.aws_availability_zones.available.names[0]
-  cidr_block              = cidrsubnet(aws_vpc.test.cidr_block, 8, 0)
-  map_public_ip_on_launch = %[1]t
-  vpc_id                  = aws_vpc.test.id
-
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
-  }
-}
-
-resource "aws_route_table" "test" {
-  vpc_id = aws_vpc.test.id
-
-  route {
-    cidr_block = "0.0.0.0/0"
-    gateway_id = aws_internet_gateway.test.id
-  }
-}
-
-resource "aws_route_table_association" "test" {
-  route_table_id = aws_route_table.test.id
-  subnet_id      = aws_subnet.test.id
-}
-`, mapPublicIPOnLaunch)
-}
-
-func testAccAWSEmrClusterConfigIAMServiceRoleBase(r string) string {
-	return fmt.Sprintf(`
-resource "aws_iam_role" "emr_service" {
-  name = "%[1]s_default_role"
-
-  assume_role_policy = <<EOT
-{
-  "Version": "2008-10-17",
-  "Statement": [
-    {
-      "Sid": "",
-      "Effect": "Allow",
-      "Principal": {
-        "Service": "elasticmapreduce.${data.aws_partition.current.dns_suffix}"
-      },
-      "Action": "sts:AssumeRole"
-    }
-  ]
-}
-EOT
-}
-
-resource "aws_iam_role_policy_attachment" "emr_service" {
-  role       = aws_iam_role.emr_service.id
-  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
-}
-
-`, r)
-}
-
-func testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(r string) string {
-	return fmt.Sprintf(`
-resource "aws_iam_role" "emr_service" {
-  name = "%[1]s_default_role"
-
-  assume_role_policy = <<EOT
-{
-  "Version": "2008-10-17",
-  "Statement": [
-    {
-      "Sid": "",
-      "Effect": "Allow",
-      "Principal": {
-        "Service": "elasticmapreduce.${data.aws_partition.current.dns_suffix}"
-      },
-      "Action": "sts:AssumeRole"
-    }
-  ]
-}
-EOT
-}
-
-resource "aws_iam_role_policy_attachment" "emr_service" {
-  role       = aws_iam_role.emr_service.id
-  policy_arn = aws_iam_policy.emr_service.arn
-}
-
-resource "aws_iam_policy" "emr_service" {
-  name = "%[1]s_emr"
-
-  policy = <<EOT
-{
-    "Version": "2012-10-17",
-    "Statement": [{
-        "Effect": "Allow",
-        "Resource": "*",
-        "Action": [
-            "ec2:AuthorizeSecurityGroupEgress",
-            "ec2:AuthorizeSecurityGroupIngress",
-            "ec2:CancelSpotInstanceRequests",
-            "ec2:CreateNetworkInterface",
-            "ec2:CreateSecurityGroup",
-            "ec2:CreateTags",
-            "ec2:DeleteNetworkInterface",
-            "ec2:DeleteSecurityGroup",
-            "ec2:DeleteTags",
-            "ec2:DescribeAvailabilityZones",
-            "ec2:DescribeAccountAttributes",
-            "ec2:DescribeDhcpOptions",
-            "ec2:DescribeImages",
-            "ec2:DescribeInstanceStatus",
-            "ec2:DescribeInstances",
-            "ec2:DescribeKeyPairs",
-            "ec2:DescribeNetworkAcls",
-            "ec2:DescribeNetworkInterfaces",
-            "ec2:DescribePrefixLists",
-            "ec2:DescribeRouteTables",
-            "ec2:DescribeSecurityGroups",
-            "ec2:DescribeSpotInstanceRequests",
-            "ec2:DescribeSpotPriceHistory",
-            "ec2:DescribeSubnets",
-            "ec2:DescribeVpcAttribute",
-            "ec2:DescribeVpcEndpoints",
-            "ec2:DescribeVpcEndpointServices",
-            "ec2:DescribeVpcs",
-            "ec2:DetachNetworkInterface",
-            "ec2:ModifyImageAttribute",
-            "ec2:ModifyInstanceAttribute",
-            "ec2:RequestSpotInstances",
-            "ec2:RevokeSecurityGroupEgress",
-            "ec2:RunInstances",
-            "ec2:TerminateInstances",
-            "ec2:DeleteVolume",
-            "ec2:DescribeVolumeStatus",
-            "ec2:DescribeVolumes",
-            "ec2:DetachVolume",
-            "iam:GetRole",
-            "iam:GetRolePolicy",
-            "iam:ListInstanceProfiles",
-            "iam:ListRolePolicies",
-            "iam:PassRole",
-            "s3:CreateBucket",
-            "s3:Get*",
-            "s3:List*",
-            "sdb:BatchPutAttributes",
-            "sdb:Select",
-            "sqs:CreateQueue",
-            "sqs:Delete*",
-            "sqs:GetQueue*",
-            "sqs:PurgeQueue",
-            "sqs:ReceiveMessage"
-        ]
-    }]
-}
-EOT
-}
-`, r)
-}
-
-func testAccAWSEmrClusterConfigIAMInstanceProfileBase(r string) string {
-	return fmt.Sprintf(`
-resource "aws_iam_instance_profile" "emr_instance_profile" {
-  name = "%[1]s_profile"
-  role = aws_iam_role.emr_instance_profile.name
-}
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
 
-resource "aws_iam_role" "emr_instance_profile" {
-  name = "%[1]s_profile_role"
+  tags = {
+    dns_zone = "new_zone"
+    Env      = "production"
+    name     = "name-env"
+  }
 
-  assume_role_policy = <<EOT
-{
-  "Version": "2008-10-17",
-  "Statement": [
-    {
-      "Sid": "",
-      "Effect": "Allow",
-      "Principal": {
-        "Service": "ec2.${data.aws_partition.current.dns_suffix}"
-      },
-      "Action": "sts:AssumeRole"
-    }
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
+
+  configurations = "test-fixtures/emr_configurations.json"
+
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+    aws_iam_role_policy_attachment.emr_autoscaling_role,
   ]
+
+  service_role     = aws_iam_role.emr_service.arn
+  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
 }
-EOT
+`, rName))
 }
 
-resource "aws_iam_role_policy_attachment" "emr_instance_profile" {
-  role       = aws_iam_role.emr_instance_profile.id
-  policy_arn = aws_iam_policy.emr_instance_profile.arn
-}
+func testAccAWSEmrClusterConfigUpdatedRootVolumeSize(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
 
-resource "aws_iam_policy" "emr_instance_profile" {
-  name = "%[1]s_profile"
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-4.6.0"
+  applications  = ["Spark"]
 
-  policy = <<EOT
-{
-    "Version": "2012-10-17",
-    "Statement": [{
-        "Effect": "Allow",
-        "Resource": "*",
-        "Action": [
-            "cloudwatch:*",
-            "dynamodb:*",
-            "ec2:Describe*",
-            "elasticmapreduce:Describe*",
-            "elasticmapreduce:ListBootstrapActions",
-            "elasticmapreduce:ListClusters",
-            "elasticmapreduce:ListInstanceGroups",
-            "elasticmapreduce:ListInstances",
-            "elasticmapreduce:ListSteps",
-            "kinesis:CreateStream",
-            "kinesis:DeleteStream",
-            "kinesis:DescribeStream",
-            "kinesis:GetRecords",
-            "kinesis:GetShardIterator",
-            "kinesis:MergeShards",
-            "kinesis:PutRecord",
-            "kinesis:SplitShard",
-            "rds:Describe*",
-            "s3:*",
-            "sdb:*",
-            "sns:*",
-            "sqs:*"
-        ]
-    }]
-}
-EOT
-}
-`, r)
-}
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  }
 
-func testAccAWSEmrClusterConfigIAMAutoscalingRole(r string) string {
-	return fmt.Sprintf(`
-resource "aws_iam_role" "emr_autoscaling_role" {
-  name               = "%[1]s_autoscaling_role"
-  assume_role_policy = data.aws_iam_policy_document.emr_autoscaling_role.json
-}
+  master_instance_group {
+    instance_type = "c4.large"
+  }
 
-data "aws_iam_policy_document" "emr_autoscaling_role" {
-  statement {
-    effect  = "Allow"
-    actions = ["sts:AssumeRole"]
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
 
-    principals {
-      type        = "Service"
-      identifiers = ["elasticmapreduce.${data.aws_partition.current.dns_suffix}", "application-autoscaling.${data.aws_partition.current.dns_suffix}"]
-    }
+  tags = {
+    role     = "rolename"
+    dns_zone = "env_zone"
+    env      = "env"
+    name     = "name-env"
   }
-}
 
-resource "aws_iam_role_policy_attachment" "emr_autoscaling_role" {
-  role       = aws_iam_role.emr_autoscaling_role.name
-  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceforAutoScalingRole"
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
+
+  configurations = "test-fixtures/emr_configurations.json"
+
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+    aws_iam_role_policy_attachment.emr_autoscaling_role,
+  ]
+
+  service_role         = aws_iam_role.emr_service.arn
+  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
+  ebs_root_volume_size = 48
 }
-`, r)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigBootstrapActionBucket(r string) string {
-	return fmt.Sprintf(`
-resource "aws_s3_bucket" "tester" {
-  bucket = "%[1]s"
-  acl    = "public-read"
-}
+func testAccAWSEmrClusterConfigStepConcurrencyLevel(rName string, stepConcurrencyLevel int) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
+resource "aws_emr_cluster" "test" {
+  applications                      = ["Spark"]
+  keep_job_flow_alive_when_no_steps = true
+  name                              = %[1]q
+  release_label                     = "emr-5.28.0"
+  service_role                      = "EMR_DefaultRole"
 
-resource "aws_s3_bucket_object" "testobject" {
-  bucket  = aws_s3_bucket.tester.bucket
-  key     = "testscript.sh"
-  content = <<EOF
-#!/bin/bash
-echo $@
-EOF
+  ec2_attributes {
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = "EMR_EC2_DefaultRole"
+    subnet_id                         = aws_subnet.test.id
+  }
 
+  master_instance_group {
+    instance_type = "m4.large"
+  }
 
-  acl = "public-read"
+  step_concurrency_level = %[2]d
+
+  depends_on = [aws_route_table_association.test]
 }
-`, r)
+`, rName, stepConcurrencyLevel))
 }
 
 func testAccAWSEmrEbsConfig(rName string, volumesPerInstance int) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		fmt.Sprintf(`
+resource "aws_emr_cluster" "test" {
   applications                      = ["Spark"]
   keep_job_flow_alive_when_no_steps = true
-  name                              = "%[1]s"
+  name                              = %[1]q
   release_label                     = "emr-5.28.0"
   service_role                      = "EMR_DefaultRole"
 
@@ -3630,19 +3576,102 @@ resource "aws_emr_cluster" "tf-test-cluster" {
 
   depends_on = [aws_route_table_association.test]
 }
-`, rName, volumesPerInstance),
-	)
+`, rName, volumesPerInstance))
+}
+
+func testAccAWSEmrClusterConfigCustomAmiID(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-5.7.0"
+  applications  = ["Spark"]
+
+  ec2_attributes {
+    subnet_id                         = aws_subnet.test.id
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+  }
+
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
+
+  tags = {
+    role     = "rolename"
+    dns_zone = "env_zone"
+    env      = "env"
+    name     = "name-env"
+  }
+
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
+
+  configurations = "test-fixtures/emr_configurations.json"
+
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+    aws_iam_role_policy_attachment.emr_autoscaling_role,
+  ]
+
+  service_role         = aws_iam_role.emr_service.arn
+  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
+  ebs_root_volume_size = 48
+  custom_ami_id        = data.aws_ami.emr-custom-ami.id
+}
+
+data "aws_ami" "emr-custom-ami" {
+  most_recent = true
+  owners      = ["amazon"]
+
+  filter {
+    name   = "name"
+    values = ["amzn-ami-hvm-*"]
+  }
+
+  filter {
+    name   = "architecture"
+    values = ["x86_64"]
+  }
+
+  filter {
+    name   = "root-device-type"
+    values = ["ebs"]
+  }
+
+  filter {
+    name   = "virtualization-type"
+    values = ["hvm"]
+  }
+}
+`, rName))
 }
 
-func testAccAWSEmrClusterConfig_InstanceFleets(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigBootstrapActionBucket(r),
+func testAccAWSEmrClusterConfig_InstanceFleets(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigBootstrapActionBucket(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-5.30.1"
   applications  = ["Hadoop", "Hive"]
   log_uri       = "s3n://terraform/testlog/"
@@ -3717,19 +3746,20 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     args = ["instance.isMaster=true", "echo running on master node"]
   }
 }
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfig_InstanceFleet_MultipleSubnets(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigBootstrapActionBucket(r),
+func testAccAWSEmrClusterConfig_InstanceFleet_MultipleSubnets(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigBootstrapActionBucket(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-5.30.1"
   applications  = ["Hadoop", "Hive"]
   log_uri       = "s3n://terraform/testlog/"
@@ -3816,19 +3846,20 @@ resource "aws_route_table_association" "test2" {
   route_table_id = aws_route_table.test.id
   subnet_id      = aws_subnet.test2.id
 }
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigInstanceFleetsMasterOnly(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigBootstrapActionBucket(r),
+func testAccAWSEmrClusterConfigInstanceFleetsMasterOnly(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigBootstrapActionBucket(rName),
 		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
   release_label = "emr-5.30.1"
   applications  = ["Hadoop", "Hive"]
   log_uri       = "s3n://terraform/testlog/"
@@ -3860,6 +3891,5 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     args = ["instance.isMaster=true", "echo running on master node"]
   }
 }
-`, r),
-	)
+`, rName))
 }