diff --git a/.changelog/17706.txt b/.changelog/17706.txt
new file mode 100644
index 00000000000..b2a8ef7639e
--- /dev/null
+++ b/.changelog/17706.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_emr_cluster: Add `log_encryption_kms_key_id` argument
+```
\ No newline at end of file
diff --git a/aws/resource_aws_emr_cluster.go b/aws/resource_aws_emr_cluster.go
index 48cc18a49b8..618b99a99f3 100644
--- a/aws/resource_aws_emr_cluster.go
+++ b/aws/resource_aws_emr_cluster.go
@@ -65,6 +65,11 @@ func resourceAwsEMRCluster() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
+			"log_encryption_kms_key_id": {
+				Type:     schema.TypeString,
+				ForceNew: true,
+				Optional: true,
+			},
 			"log_uri": {
 				Type:     schema.TypeString,
 				ForceNew: true,
@@ -856,6 +861,10 @@ func resourceAwsEMRClusterCreate(d *schema.ResourceData, meta interface{}) error
 		params.AdditionalInfo = aws.String(info)
 	}
 
+	if v, ok := d.GetOk("log_encryption_kms_key_id"); ok {
+		params.LogEncryptionKmsKeyId = aws.String(v.(string))
+	}
+
 	if v, ok := d.GetOk("log_uri"); ok {
 		params.LogUri = aws.String(v.(string))
 	}
@@ -1093,6 +1102,7 @@ func resourceAwsEMRClusterRead(d *schema.ResourceData, meta interface{}) error {
 	d.Set("security_configuration", cluster.SecurityConfiguration)
 	d.Set("autoscaling_role", cluster.AutoScalingRole)
 	d.Set("release_label", cluster.ReleaseLabel)
+	d.Set("log_encryption_kms_key_id", cluster.LogEncryptionKmsKeyId)
 	d.Set("log_uri", cluster.LogUri)
 	d.Set("master_public_dns", cluster.MasterPublicDnsName)
 	d.Set("visible_to_all_users", cluster.VisibleToAllUsers)
diff --git a/aws/resource_aws_emr_cluster_test.go b/aws/resource_aws_emr_cluster_test.go
index a10cca67782..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{
@@ -1201,10 +1201,44 @@ func TestAccAWSEMRCluster_s3Logging(t *testing.T) {
 	})
 }
 
+func TestAccAWSEMRCluster_s3LogEncryption(t *testing.T) {
+	var cluster emr.Cluster
+
+	resourceName := "aws_emr_cluster.test"
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	bucketName := fmt.Sprintf("s3n://%s/", rName)
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		ErrorCheck:   testAccErrorCheck(t, emr.EndpointsID),
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSEmrDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccAWSEmrClusterConfigS3Encryption(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSEmrClusterExists(resourceName, &cluster),
+					resource.TestCheckResourceAttr(resourceName, "log_uri", bucketName),
+					testAccMatchResourceAttrRegionalARN(resourceName, "log_encryption_kms_key_id", "kms", regexp.MustCompile(`key/.+`)),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"cluster_state", // Ignore RUNNING versus WAITING changes
+					"configurations",
+					"keep_job_flow_alive_when_no_steps",
+				},
+			},
+		},
+	})
+}
+
 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) },
@@ -1249,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) },
@@ -1288,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),
@@ -1326,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),
@@ -1358,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) },
@@ -1390,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")
@@ -1457,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) },
@@ -1697,235 +1731,252 @@ 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"
-  }
+// Sub-configs (used by other configs)
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
+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"]
   }
+}
 
-  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
-  }
+resource "aws_vpc" "test" {
+  cidr_block           = "10.0.0.0/16"
+  enable_dns_hostnames = true
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif"
-    args = ["instance.isMaster=true", "echo running on master node"]
+  tags = {
+    Name = %[1]q
   }
+}
 
-  bootstrap_action {
-    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
-    name = "test"
+resource "aws_internet_gateway" "test" {
+  vpc_id = aws_vpc.test.id
 
-    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/"
+resource "aws_security_group" "test" {
+  vpc_id = aws_vpc.test.id
 
-  master_instance_group {
-    instance_type = "c4.large"
+  ingress {
+    from_port = 0
+    protocol  = "-1"
+    self      = true
+    to_port   = 0
   }
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
+  egress {
+    cidr_blocks = ["0.0.0.0/0"]
+    from_port   = 0
+    protocol    = "-1"
+    to_port     = 0
   }
 
-  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
+  tags = {
+    Name = %[1]q
   }
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif"
-    args = ["instance.isMaster=true", "echo running on master node"]
+  # EMR will modify ingress rules
+  lifecycle {
+    ignore_changes = [ingress]
   }
+}
 
-  bootstrap_action {
-    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
-    name = "test"
+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
 
-    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"]
+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
   }
 }
-`, 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/"
+resource "aws_route_table_association" "test" {
+  route_table_id = aws_route_table.test.id
+  subnet_id      = aws_subnet.test.id
+}
+`, rName, mapPublicIPOnLaunch)
+}
 
-  master_instance_group {
-    instance_type = "c4.large"
-  }
+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
+}
 
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
+resource "aws_iam_role" "emr_instance_profile" {
+  name = "%[1]s_profile_role"
 
-  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,
+  assume_role_policy = <<EOT
+{
+  "Version": "2008-10-17",
+  "Statement": [
+    {
+      "Sid": "",
+      "Effect": "Allow",
+      "Principal": {
+        "Service": "ec2.${data.aws_partition.current.dns_suffix}"
+      },
+      "Action": "sts:AssumeRole"
+    }
   ]
+}
+EOT
+}
 
-  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_role_policy_attachment" "emr_instance_profile" {
+  role       = aws_iam_role.emr_instance_profile.id
+  policy_arn = aws_iam_policy.emr_instance_profile.arn
+}
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif"
-    args = ["instance.isMaster=true", "echo running on master node"]
-  }
+resource "aws_iam_policy" "emr_instance_profile" {
+  name = "%[1]s_profile"
 
-  bootstrap_action {
-    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
-    name = "runif-2"
-    args = ["instance.isMaster=true", "echo also running on master node"]
-  }
+  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)
+}
 
-  bootstrap_action {
-    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
-    name = "test"
+func testAccAWSEmrClusterConfigIAMServiceRoleBase(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_iam_role" "emr_service" {
+  name = "%[1]s_default_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": "elasticmapreduce.${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_service" {
+  role       = aws_iam_role.emr_service.id
+  policy_arn = "arn:${data.aws_partition.current.partition}:iam::aws:policy/service-role/AmazonElasticMapReduceRole"
+}
 
-  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)
+}
+
+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
+}
+
+data "aws_iam_policy_document" "emr_autoscaling_role" {
+  statement {
+    effect  = "Allow"
+    actions = ["sts:AssumeRole"]
+
+    principals {
+      type        = "Service"
+      identifiers = ["elasticmapreduce.${data.aws_partition.current.dns_suffix}", "application-autoscaling.${data.aws_partition.current.dns_suffix}"]
+    }
   }
+}
+
+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"
+}
+`, rName)
+}
+
+func testAccAWSEmrClusterConfigBootstrapActionBucket(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_s3_bucket" "tester" {
+  bucket = %[1]q
+  acl    = "public-read"
+}
+
+resource "aws_s3_bucket_object" "testobject" {
+  bucket  = aws_s3_bucket.tester.bucket
+  key     = "testscript.sh"
+  content = <<EOF
+#!/bin/bash
+echo $@
+EOF
+
+
+  acl = "public-read"
+}
+`, rName)
+}
+
+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
+  log_uri                           = "s3://${aws_s3_bucket.test.bucket}/"
+  name                              = %[1]q
+  release_label                     = "emr-5.12.0"
+  service_role                      = "EMR_DefaultRole"
+  termination_protection            = false
 
   master_instance_group {
     instance_type = "c4.large"
@@ -1936,56 +1987,166 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
+  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
   }
 
-  keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
+  %[2]s
 
-  scale_down_behavior = "TERMINATE_AT_TASK_COMPLETION"
+  depends_on = [aws_route_table_association.test]
+}
 
-  configurations = "test-fixtures/emr_configurations.json"
+resource "aws_s3_bucket" "test" {
+  bucket        = %[1]q
+  force_destroy = true
+}
+`, rName, stepConfig))
+}
 
-  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,
+func testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(rName 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
+}
 
-  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_service" {
+  role       = aws_iam_role.emr_service.id
+  policy_arn = aws_iam_policy.emr_service.arn
 }
-`, r),
-	)
+
+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
+}
+`, rName)
 }
 
-func testAccAWSEmrClusterConfigAdditionalInfo(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+const testAccAWSEmrClusterConfig_Step_DebugLoggingStep = `
+  # Example from: https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-debugging.html
+  step {
+    action_on_failure = "TERMINATE_CLUSTER"
+    name              = "Setup Hadoop Debugging"
+
+    hadoop_jar_step {
+      jar  = "command-runner.jar"
+      args = ["state-pusher-script"]
+    }
+  }
+`
+
+const testAccAWSEmrClusterConfig_Step_SparkStep = `
+  # Example from: https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-submit-step.html
+  step {
+    action_on_failure = "CONTINUE"
+    name              = "Spark Step"
+
+    hadoop_jar_step {
+      jar  = "command-runner.jar"
+      args = ["spark-example", "SparkPi", "10"]
+    }
+  }
+`
+
+// Configs
+
+func testAccAWSEmrClusterConfig(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"]
 
-  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
@@ -2027,145 +2188,31 @@ EOF
   autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
   ebs_root_volume_size = 21
 }
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigConfigurationsJson(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
+func testAccAWSEmrClusterConfigAdditionalInfo(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"
-  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,
-  ]
+data "aws_partition" "current" {}
 
-  service_role         = aws_iam_role.emr_service.arn
-  ebs_root_volume_size = 21
-}
-`, r),
-	)
-}
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-4.6.0"
+  applications  = ["Spark"]
 
-func testAccAWSEmrClusterConfig_Kerberos_ClusterDedicatedKdc(r string, password string) string {
-	return testAccAWSEmrComposeConfig(false, fmt.Sprintf(`
-resource "aws_emr_security_configuration" "foo" {
-  configuration = <<EOF
+  additional_info = <<EOF
 {
-  "AuthenticationConfiguration": {
-    "KerberosConfiguration": {
-      "Provider": "ClusterDedicatedKdc",
-      "ClusterDedicatedKdcConfiguration": {
-        "TicketLifetimeInHours": 24
-      }
-    }
-  }
+	"instanceAwsClientConfiguration": {
+		"proxyPort": 8099,
+		"proxyHost": "myproxy.example.com"
+	}
 }
 EOF
-}
-
-resource "aws_emr_cluster" "tf-test-cluster" {
-  applications                      = ["Spark"]
-  keep_job_flow_alive_when_no_steps = true
-  name                              = "%[1]s"
-  release_label                     = "emr-5.12.0"
-  security_configuration            = aws_emr_security_configuration.foo.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]
-}
-`, 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"]
 
   ec2_attributes {
     subnet_id                         = aws_subnet.test.id
@@ -2183,8 +2230,6 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
-  security_configuration = aws_emr_security_configuration.foo.name
-
   tags = {
     role     = "rolename"
     dns_zone = "env_zone"
@@ -2195,117 +2240,43 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   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
-{
-  "Version": "2012-10-17",
-  "Id": "kms-tf-1",
-  "Statement": [
-    {
-      "Sid": "Enable IAM User Permissions",
-      "Effect": "Allow",
-      "Principal": {
-        "AWS": "*"
-      },
-      "Action": "kms:*",
-      "Resource": "*"
-    }
-  ]
-}
-POLICY
-}
-`, r))
-}
-
-const testAccAWSEmrClusterConfig_Step_DebugLoggingStep = `
-  # Example from: https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-debugging.html
-  step {
-    action_on_failure = "TERMINATE_CLUSTER"
-    name              = "Setup Hadoop Debugging"
-
-    hadoop_jar_step {
-      jar  = "command-runner.jar"
-      args = ["state-pusher-script"]
-    }
-  }
-`
-
-const testAccAWSEmrClusterConfig_Step_SparkStep = `
-  # Example from: https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-submit-step.html
-  step {
-    action_on_failure = "CONTINUE"
-    name              = "Spark Step"
-
-    hadoop_jar_step {
-      jar  = "command-runner.jar"
-      args = ["spark-example", "SparkPi", "10"]
-    }
-  }
-`
-
-func testAccAWSEmrClusterConfig_Step_Multiple(r string) string {
-	stepConfig := testAccAWSEmrClusterConfig_Step_DebugLoggingStep + testAccAWSEmrClusterConfig_Step_SparkStep
-	return testAccAWSEmrClusterConfig_Step(r, stepConfig)
-}
+  scale_down_behavior = "TERMINATE_AT_TASK_COMPLETION"
 
-func testAccAWSEmrClusterConfig_Step_NoBlocks(r string) string {
-	return testAccAWSEmrClusterConfig_Step(r, "")
-}
+  configurations = "test-fixtures/emr_configurations.json"
 
-func testAccAWSEmrClusterConfig_Step_Single(r string) string {
-	return testAccAWSEmrClusterConfig_Step(r, testAccAWSEmrClusterConfig_Step_DebugLoggingStep)
-}
+  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,
+  ]
 
-func testAccAWSEmrClusterConfig_Step_Zeroed(r string) string {
-	return testAccAWSEmrClusterConfig_Step(r, "step = []")
+  service_role         = aws_iam_role.emr_service.arn
+  autoscaling_role     = aws_iam_role.emr_autoscaling_role.arn
+  ebs_root_volume_size = 21
+}
+`, rName))
 }
 
-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
+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"
@@ -2316,29 +2287,56 @@ 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
-  }
+  keep_job_flow_alive_when_no_steps = true
+  termination_protection            = false
 
-  %[2]s
+  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]
-}
+  depends_on = [
+    aws_route_table_association.test,
+    aws_iam_role_policy_attachment.emr_service,
+    aws_iam_role_policy_attachment.emr_instance_profile,
+  ]
 
-resource "aws_s3_bucket" "test" {
-  bucket        = "%[1]s"
-  force_destroy = true
+  service_role         = aws_iam_role.emr_service.arn
+  ebs_root_volume_size = 21
 }
-`, r, stepConfig))
+`, rName))
 }
 
 func testAccAWSEmrClusterConfigCoreInstanceGroupAutoscalingPolicy(rName, autoscalingPolicy string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
 		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
 data "aws_iam_policy_document" "test" {
   statement {
     actions = ["sts:AssumeRole"]
@@ -2395,14 +2393,15 @@ POLICY
     aws_iam_role_policy_attachment.test,
   ]
 }
-`, rName, autoscalingPolicy),
-	)
+`, rName, autoscalingPolicy))
 }
 
 func testAccAWSEmrClusterConfigCoreInstanceGroupAutoscalingPolicyRemoved(rName string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
 		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
 data "aws_iam_policy_document" "test" {
   statement {
     actions = ["sts:AssumeRole"]
@@ -2456,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
@@ -2487,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
@@ -2518,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
@@ -2548,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
@@ -2579,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
@@ -2603,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
@@ -2630,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
@@ -2665,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
@@ -2691,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
@@ -2718,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 {
@@ -2750,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"
@@ -2758,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"
 
@@ -2772,21 +2837,99 @@ 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),
-	)
+
+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
+}
+
+resource "aws_kms_key" "test" {
+  description             = "Terraform %[1]s"
+  deletion_window_in_days = 7
+
+  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_keepJob(r string, keepJob string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleBase(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
+func testAccAWSEmrClusterConfig_bootstrap(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"
-  release_label = "emr-4.6.0"
-  applications  = ["Spark"]
+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
@@ -2795,6 +2938,47 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     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",
+    ]
+  }
+}
+`, rName))
+}
+
+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"
   }
@@ -2804,54 +2988,13 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
-  }
-
-  keep_job_flow_alive_when_no_steps = %s
-  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"
-
+  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
-}
-`, r, keepJob),
-	)
-}
-
-func testAccAWSEmrClusterConfigVisibleToAllUsersUpdated(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"]
-
   ec2_attributes {
     subnet_id                         = aws_subnet.test.id
     emr_managed_master_security_group = aws_security_group.test.id
@@ -2859,6 +3002,53 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     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"
   }
@@ -2868,44 +3058,13 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     instance_type  = "c4.large"
   }
 
-  tags = {
-    role     = "rolename"
-    dns_zone = "env_zone"
-    env      = "env"
-    name     = "name-env"
-  }
-
-  keep_job_flow_alive_when_no_steps = true
-  visible_to_all_users              = false
-
-  configurations = "test-fixtures/emr_configurations.json"
-
+  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
-}
-`, r),
-	)
-}
-
-func testAccAWSEmrClusterConfigUpdatedTags(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"]
-
   ec2_attributes {
     subnet_id                         = aws_subnet.test.id
     emr_managed_master_security_group = aws_security_group.test.id
@@ -2913,49 +3072,49 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     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"
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif"
+    args = ["instance.isMaster=true", "echo running on master node"]
   }
 
-  tags = {
-    dns_zone = "new_zone"
-    Env      = "production"
-    name     = "name-env"
+  bootstrap_action {
+    path = "s3://elasticmapreduce/bootstrap-actions/run-if"
+    name = "runif-2"
+    args = ["instance.isMaster=true", "echo also running on master node"]
   }
 
-  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,
-  ]
+  bootstrap_action {
+    path = "s3://${aws_s3_bucket_object.testobject.bucket}/${aws_s3_bucket_object.testobject.key}"
+    name = "test"
 
-  service_role     = aws_iam_role.emr_service.arn
-  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
+    args = ["1",
+      "2",
+      "3",
+      "4",
+      "5",
+      "6",
+      "7",
+      "8",
+      "9",
+      "10",
+    ]
+  }
 }
-`, r),
-	)
+`, rName))
 }
 
-func testAccAWSEmrClusterConfigUpdatedRootVolumeSize(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"]
 
@@ -2983,7 +3142,7 @@ resource "aws_emr_cluster" "tf-test-cluster" {
   }
 
   keep_job_flow_alive_when_no_steps = true
-  termination_protection            = false
+  termination_protection            = %[2]s
 
   configurations = "test-fixtures/emr_configurations.json"
 
@@ -2994,69 +3153,26 @@ 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, term))
 }
 
-func testAccAWSEmrClusterConfigS3Logging(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
+func testAccAWSEmrClusterConfig_keepJob(rName string, keepJob bool) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleBase(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		testAccAWSEmrClusterConfigIAMAutoscalingRole(rName),
 		fmt.Sprintf(`
-resource "aws_s3_bucket" "test" {
-  bucket        = "%[1]s"
-  force_destroy = true
-}
+data "aws_partition" "current" {}
 
-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"]
 
-  termination_protection            = false
-  keep_job_flow_alive_when_no_steps = true
-
-  master_instance_group {
-    instance_type = "c4.large"
-  }
-
-  core_instance_group {
-    instance_count = 1
-    instance_type  = "c4.large"
-  }
-
-  log_uri = "s3://${aws_s3_bucket.test.bucket}/"
-
-  ec2_attributes {
-    instance_profile                  = "arn:${data.aws_partition.current.partition}:iam::${data.aws_caller_identity.current.account_id}:instance-profile/EMR_EC2_DefaultRole"
-    emr_managed_master_security_group = aws_security_group.test.id
-    emr_managed_slave_security_group  = aws_security_group.test.id
-    subnet_id                         = aws_subnet.test.id
-  }
-
-  service_role = "arn:${data.aws_partition.current.partition}:iam::${data.aws_caller_identity.current.account_id}:role/EMR_DefaultRole"
-}
-
-data "aws_caller_identity" "current" {}
-`, r),
-	)
-}
-
-func testAccAWSEmrClusterConfigCustomAmiID(r string) string {
-	return testAccAWSEmrComposeConfig(false,
-		testAccAWSEmrClusterConfigCurrentPartition(),
-		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(r),
-		testAccAWSEmrClusterConfigIAMInstanceProfileBase(r),
-		testAccAWSEmrClusterConfigIAMAutoscalingRole(r),
-		fmt.Sprintf(`
-resource "aws_emr_cluster" "tf-test-cluster" {
-  name          = "%[1]s"
-  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
@@ -3080,9 +3196,19 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     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 = [
@@ -3092,405 +3218,329 @@ 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
-  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"]
-  }
+  service_role     = aws_iam_role.emr_service.arn
+  autoscaling_role = aws_iam_role.emr_autoscaling_role.arn
 }
-`, r),
-	)
+`, rName, keepJob))
 }
 
-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"
+func testAccAWSEmrClusterConfigVisibleToAllUsersUpdated(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" {
+  name          = %[1]q
+  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                  = "EMR_EC2_DefaultRole"
-    subnet_id                         = aws_subnet.test.id
-  }
-
-  master_instance_group {
-    instance_type = "m4.large"
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
   }
 
-  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"]
+  master_instance_group {
+    instance_type = "c4.large"
   }
-}
 
-resource "aws_vpc" "test" {
-  cidr_block           = "10.0.0.0/16"
-  enable_dns_hostnames = true
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
 
   tags = {
-    Name = "tf-acc-test-emr-cluster"
+    role     = "rolename"
+    dns_zone = "env_zone"
+    env      = "env"
+    name     = "name-env"
   }
-}
 
-resource "aws_internet_gateway" "test" {
-  vpc_id = aws_vpc.test.id
+  keep_job_flow_alive_when_no_steps = true
+  visible_to_all_users              = false
 
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
-  }
+  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
+}
+`, rName))
 }
 
-resource "aws_security_group" "test" {
-  vpc_id = aws_vpc.test.id
+func testAccAWSEmrClusterConfigS3Logging(rName string) string {
+	return composeConfig(
+		testAccAWSEmrClusterConfigBaseVpc(rName, false),
+		testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(rName),
+		testAccAWSEmrClusterConfigIAMInstanceProfileBase(rName),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
 
-  ingress {
-    from_port = 0
-    protocol  = "-1"
-    self      = true
-    to_port   = 0
-  }
+resource "aws_s3_bucket" "test" {
+  bucket        = %[1]q
+  force_destroy = true
+}
 
-  egress {
-    cidr_blocks = ["0.0.0.0/0"]
-    from_port   = 0
-    protocol    = "-1"
-    to_port     = 0
-  }
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-4.6.0"
+  applications  = ["Spark"]
 
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
-  }
+  termination_protection            = false
+  keep_job_flow_alive_when_no_steps = true
 
-  # EMR will modify ingress rules
-  lifecycle {
-    ignore_changes = [ingress]
+  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 = %[1]t
-  vpc_id                  = aws_vpc.test.id
 
-  tags = {
-    Name = "tf-acc-test-emr-cluster"
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
   }
-}
 
-resource "aws_route_table" "test" {
-  vpc_id = aws_vpc.test.id
+  log_uri = "s3://${aws_s3_bucket.test.bucket}/"
 
-  route {
-    cidr_block = "0.0.0.0/0"
-    gateway_id = aws_internet_gateway.test.id
+  ec2_attributes {
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    subnet_id                         = aws_subnet.test.id
   }
+
+  service_role = aws_iam_role.emr_service.arn
 }
 
-resource "aws_route_table_association" "test" {
-  route_table_id = aws_route_table.test.id
-  subnet_id      = aws_subnet.test.id
+data "aws_caller_identity" "current" {}
+`, rName))
 }
-`, mapPublicIPOnLaunch)
+
+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]q
+  force_destroy = true
 }
 
-func testAccAWSEmrClusterConfigIAMServiceRoleBase(r string) string {
-	return fmt.Sprintf(`
-resource "aws_iam_role" "emr_service" {
-  name = "%[1]s_default_role"
+resource "aws_kms_key" "test" {
+  description = "Terraform acc test %[1]s"
 
-  assume_role_policy = <<EOT
+  policy = <<POLICY
 {
-  "Version": "2008-10-17",
+  "Version": "2012-10-17",
+  "Id": "kms-tf-1",
   "Statement": [
     {
-      "Sid": "",
+      "Sid": "Enable IAM User Permissions",
       "Effect": "Allow",
       "Principal": {
-        "Service": "elasticmapreduce.${data.aws_partition.current.dns_suffix}"
+        "AWS": "*"
       },
-      "Action": "sts:AssumeRole"
+      "Action": "kms:*",
+      "Resource": "*"
     }
   ]
 }
-EOT
+POLICY
 }
 
-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"
-}
+resource "aws_emr_cluster" "test" {
+  name          = %[1]q
+  release_label = "emr-5.32.0"
+  applications  = ["Spark"]
 
-`, r)
-}
+  termination_protection            = false
+  keep_job_flow_alive_when_no_steps = true
 
-func testAccAWSEmrClusterConfigIAMServiceRoleCustomAmiID(r string) string {
-	return fmt.Sprintf(`
-resource "aws_iam_role" "emr_service" {
-  name = "%[1]s_default_role"
+  master_instance_group {
+    instance_type = "c4.large"
+  }
 
-  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
-}
+  core_instance_group {
+    instance_count = 1
+    instance_type  = "c4.large"
+  }
 
-resource "aws_iam_role_policy_attachment" "emr_service" {
-  role       = aws_iam_role.emr_service.id
-  policy_arn = aws_iam_policy.emr_service.arn
-}
+  log_encryption_kms_key_id = aws_kms_key.test.arn
+  log_uri                   = "s3://${aws_s3_bucket.test.bucket}/"
 
-resource "aws_iam_policy" "emr_service" {
-  name = "%[1]s_emr"
+  ec2_attributes {
+    instance_profile                  = aws_iam_instance_profile.emr_instance_profile.arn
+    emr_managed_master_security_group = aws_security_group.test.id
+    emr_managed_slave_security_group  = aws_security_group.test.id
+    subnet_id                         = aws_subnet.test.id
+  }
 
-  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
+  service_role = aws_iam_role.emr_service.arn
 }
-`, r)
+
+data "aws_caller_identity" "current" {}
+`, rName))
 }
 
-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
-}
+func testAccAWSEmrClusterConfigUpdatedTags(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" {
+  name          = %[1]q
+  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
+  }
+
+  master_instance_group {
+    instance_type = "c4.large"
+  }
+
+  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"
 
@@ -3526,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/"
@@ -3613,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/"
@@ -3712,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/"
@@ -3756,6 +3891,5 @@ resource "aws_emr_cluster" "tf-test-cluster" {
     args = ["instance.isMaster=true", "echo running on master node"]
   }
 }
-`, r),
-	)
+`, rName))
 }
diff --git a/website/docs/r/emr_cluster.html.markdown b/website/docs/r/emr_cluster.html.markdown
index 45f64c50f6d..9b28f8a540f 100644
--- a/website/docs/r/emr_cluster.html.markdown
+++ b/website/docs/r/emr_cluster.html.markdown
@@ -8,9 +8,7 @@ description: |-
 
 # Resource: aws_emr_cluster
 
-Provides an Elastic MapReduce Cluster, a web service that makes it easy to
-process large amounts of data efficiently. See [Amazon Elastic MapReduce Documentation](https://aws.amazon.com/documentation/elastic-mapreduce/)
-for more information.
+Provides an Elastic MapReduce Cluster, a web service that makes it easy to process large amounts of data efficiently. See [Amazon Elastic MapReduce Documentation](https://aws.amazon.com/documentation/elastic-mapreduce/) for more information.
 
 To configure [Instance Groups](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-group-configuration.html#emr-plan-instance-groups) for [task nodes](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-master-core-task-nodes.html#emr-plan-task), see the [`aws_emr_instance_group` resource](/docs/providers/aws/r/emr_instance_group.html).
 
@@ -138,15 +136,9 @@ EOF
 }
 ```
 
-The `aws_emr_cluster` resource typically requires two IAM roles, one for the EMR Cluster
-to use as a service, and another to place on your Cluster Instances to interact
-with AWS from those instances. The suggested role policy template for the EMR service is `AmazonElasticMapReduceRole`,
-and `AmazonElasticMapReduceforEC2Role` for the EC2 profile. See the [Getting
-Started](https://docs.aws.amazon.com/ElasticMapReduce/latest/ManagementGuide/emr-gs-launch-sample-cluster.html)
-guide for more information on these IAM roles. There is also a fully-bootable
-example Terraform configuration at the bottom of this page.
+The `aws_emr_cluster` resource typically requires two IAM roles, one for the EMR Cluster to use as a service, and another to place on your Cluster Instances to interact with AWS from those instances. The suggested role policy template for the EMR service is `AmazonElasticMapReduceRole`, and `AmazonElasticMapReduceforEC2Role` for the EC2 profile. See the [Getting Started](https://docs.aws.amazon.com/ElasticMapReduce/latest/ManagementGuide/emr-gs-launch-sample-cluster.html) guide for more information on these IAM roles. There is also a fully-bootable example Terraform configuration at the bottom of this page.
 
-## Instance Fleet
+### Instance Fleet
 
 ```terraform
 resource "aws_emr_cluster" "example" {
@@ -240,10 +232,7 @@ resource "aws_emr_instance_fleet" "task" {
 
 ### Enable Debug Logging
 
-[Debug logging in EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-debugging.html)
-is implemented as a step. It is highly recommended to utilize the
-[lifecycle configuration block](https://www.terraform.io/docs/configuration/meta-arguments/lifecycle.html) with `ignore_changes` if other
-steps are being managed outside of Terraform.
+[Debug logging in EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-debugging.html) is implemented as a step. It is highly recommended that you utilize the [lifecycle configuration block](https://www.terraform.io/docs/configuration/meta-arguments/lifecycle.html) with `ignore_changes` if other steps are being managed outside of Terraform.
 
 ```terraform
 resource "aws_emr_cluster" "example" {
@@ -315,242 +304,9 @@ resource "aws_emr_cluster" "example" {
 }
 ```
 
-## Argument Reference
-
-The following arguments are supported:
-
-* `name` - (Required) The name of the job flow
-* `release_label` - (Required) The release label for the Amazon EMR release
-* `master_instance_group` - (Optional) Configuration block to use an [Instance Group](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-group-configuration.html#emr-plan-instance-groups) for the [master node type](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-master-core-task-nodes.html#emr-plan-master).
-* `master_instance_fleet` - (Optional) Configuration block to use an [Instance Fleet](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-fleet.html) for the master node type. Cannot be specified if any `master_instance_group` configuration blocks are set. Detailed below.
-* `scale_down_behavior` - (Optional) The way that individual Amazon EC2 instances terminate when an automatic scale-in activity occurs or an `instance group` is resized.
-* `additional_info` - (Optional) A JSON string for selecting additional features such as adding proxy information. Note: Currently there is no API to retrieve the value of this argument after EMR cluster creation from provider, therefore Terraform cannot detect drift from the actual EMR cluster if its value is changed outside Terraform.
-* `service_role` - (Required) IAM role that will be assumed by the Amazon EMR service to access AWS resources
-* `security_configuration` - (Optional) The security configuration name to attach to the EMR cluster. Only valid for EMR clusters with `release_label` 4.8.0 or greater
-* `core_instance_group` - (Optional) Configuration block to use an [Instance Group](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-group-configuration.html#emr-plan-instance-groups) for the [core node type](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-master-core-task-nodes.html#emr-plan-core).
-* `core_instance_fleet` - (Optional) Configuration block to use an [Instance Fleet](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-fleet.html) for the core node type. Cannot be specified if any `core_instance_group` configuration blocks are set. Detailed below.
-* `log_uri` - (Optional) S3 bucket to write the log files of the job flow. If a value is not provided, logs are not created
-* `applications` - (Optional) A list of applications for the cluster. Valid values are: `Flink`, `Hadoop`, `Hive`, `Mahout`, `Pig`, `Spark`, and `JupyterHub` (as of EMR 5.14.0). Case insensitive
-* `termination_protection` - (Optional) Switch on/off termination protection (default is `false`, except when using multiple master nodes). Before attempting to destroy the resource when termination protection is enabled, this configuration must be applied with its value set to `false`.
-* `keep_job_flow_alive_when_no_steps` - (Optional) Switch on/off run cluster with no steps or when all steps are complete (default is on)
-* `ec2_attributes` - (Optional) Attributes for the EC2 instances running the job flow. Defined below
-* `kerberos_attributes` - (Optional) Kerberos configuration for the cluster. Defined below
-* `ebs_root_volume_size` - (Optional) Size in GiB of the EBS root device volume of the Linux AMI that is used for each EC2 instance. Available in Amazon EMR version 4.x and later.
-* `custom_ami_id` - (Optional) A custom Amazon Linux AMI for the cluster (instead of an EMR-owned AMI). Available in Amazon EMR version 5.7.0 and later.
-* `bootstrap_action` - (Optional) Ordered list of bootstrap actions that will be run before Hadoop is started on the cluster nodes. Defined below.
-* `configurations` - (Optional) List of configurations supplied for the EMR cluster you are creating. Supply a configuration object for applications to override their default configuration. See [AWS Documentation](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html) for more information.
-* `configurations_json` - (Optional) A JSON string for supplying list of configurations for the EMR cluster.
-
-~> **NOTE on configurations_json:** If the `Configurations` value is empty then you should skip
-the `Configurations` field instead of providing empty list as value `"Configurations": []`.
-
-```terraform
-resource "aws_emr_cluster" "cluster" {
-  # ... other configuration ...
-
-  configurations_json = <<EOF
-  [
-    {
-      "Classification": "hadoop-env",
-      "Configurations": [
-        {
-          "Classification": "export",
-          "Properties": {
-            "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"
-          }
-        }
-      ],
-      "Properties": {}
-    }
-  ]
-EOF
-}
-```
-
-* `visible_to_all_users` - (Optional) Whether the job flow is visible to all IAM users of the AWS account associated with the job flow. Default `true`
-* `autoscaling_role` - (Optional) An IAM role for automatic scaling policies. The IAM role provides permissions that the automatic scaling feature requires to launch and terminate EC2 instances in an instance group.
-* `step` - (Optional) List of steps to run when creating the cluster. Defined below. It is highly recommended to utilize the [lifecycle configuration block](https://www.terraform.io/docs/configuration/meta-arguments/lifecycle.html) with `ignore_changes` if other steps are being managed outside of Terraform. This argument is processed in [attribute-as-blocks mode](https://www.terraform.io/docs/configuration/attr-as-blocks.html).
-* `step_concurrency_level` - (Optional) The number of steps that can be executed concurrently. You can specify a maximum of 256 steps. Only valid for EMR clusters with `release_label` 5.28.0 or greater. (default is 1)
-* `tags` - (Optional) list of tags to apply to the EMR Cluster. If configured with a provider [`default_tags` configuration block](https://www.terraform.io/docs/providers/aws/index.html#default_tags-configuration-block) present, tags with matching keys will overwrite those defined at the provider-level.
-
-## core_instance_group Configuration Block
-
-Supported arguments for the `core_instance_group` configuration block:
-
-* `instance_type` - (Required) EC2 instance type for all instances in the instance group.
-* `autoscaling_policy` - (Optional) String containing the [EMR Auto Scaling Policy](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-automatic-scaling.html) JSON.
-* `bid_price` - (Optional) Bid price for each EC2 instance in the instance group, expressed in USD. By setting this attribute, the instance group is being declared as a Spot Instance, and will implicitly create a Spot request. Leave this blank to use On-Demand Instances.
-* `ebs_config` - (Optional) Configuration block(s) for EBS volumes attached to each instance in the instance group. Detailed below.
-* `instance_count` - (Optional) Target number of instances for the instance group. Must be at least 1. Defaults to 1.
-* `name` - (Optional) Friendly name given to the instance group.
-
-## ec2_attributes
-
-Attributes for the Amazon EC2 instances running the job flow
-
-* `key_name` - (Optional) Amazon EC2 key pair that can be used to ssh to the master node as the user called `hadoop`
-* `subnet_id` - (Optional) VPC subnet id where you want the job flow to launch. Cannot specify the `cc1.4xlarge` instance type for nodes of a job flow launched in a Amazon VPC
-* `subnet_ids` - (Optional) List of VPC subnet id-s where you want the job flow to launch.  Amazon EMR identifies the best Availability Zone to launch instances according to your fleet specifications
-* `additional_master_security_groups` - (Optional) String containing a comma separated list of additional Amazon EC2 security group IDs for the master node
-* `additional_slave_security_groups` - (Optional) String containing a comma separated list of additional Amazon EC2 security group IDs for the slave nodes as a comma separated string
-* `emr_managed_master_security_group` - (Optional) Identifier of the Amazon EC2 EMR-Managed security group for the master node
-* `emr_managed_slave_security_group` - (Optional) Identifier of the Amazon EC2 EMR-Managed security group for the slave nodes
-* `service_access_security_group` - (Optional) Identifier of the Amazon EC2 service-access security group - required when the cluster runs on a private subnet
-* `instance_profile` - (Required) Instance Profile for EC2 instances of the cluster assume this role
-
-~> **NOTE on EMR-Managed security groups:** These security groups will have any
-missing inbound or outbound access rules added and maintained by AWS, to ensure
-proper communication between instances in a cluster. The EMR service will
-maintain these rules for groups provided in `emr_managed_master_security_group`
-and `emr_managed_slave_security_group`; attempts to remove the required rules
-may succeed, only for the EMR service to re-add them in a matter of minutes.
-This may cause Terraform to fail to destroy an environment that contains an EMR
-cluster, because the EMR service does not revoke rules added on deletion,
-leaving a cyclic dependency between the security groups that prevents their
-deletion. To avoid this, use the `revoke_rules_on_delete` optional attribute for
-any Security Group used in `emr_managed_master_security_group` and
-`emr_managed_slave_security_group`. See [Amazon EMR-Managed Security
-Groups](http://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-man-sec-groups.html)
-for more information about the EMR-managed security group rules.
-
-## kerberos_attributes
-
-Attributes for Kerberos configuration
-
-* `ad_domain_join_password` - (Optional) The Active Directory password for `ad_domain_join_user`. Terraform cannot perform drift detection of this configuration.
-* `ad_domain_join_user` - (Optional) Required only when establishing a cross-realm trust with an Active Directory domain. A user with sufficient privileges to join resources to the domain. Terraform cannot perform drift detection of this configuration.
-* `cross_realm_trust_principal_password` - (Optional) Required only when establishing a cross-realm trust with a KDC in a different realm. The cross-realm principal password, which must be identical across realms. Terraform cannot perform drift detection of this configuration.
-* `kdc_admin_password` - (Required) The password used within the cluster for the kadmin service on the cluster-dedicated KDC, which maintains Kerberos principals, password policies, and keytabs for the cluster. Terraform cannot perform drift detection of this configuration.
-* `realm` - (Required) The name of the Kerberos realm to which all nodes in a cluster belong. For example, `EC2.INTERNAL`
-
-## master_instance_group Configuration Block
-
-Supported nested arguments for the `master_instance_group` configuration block:
-
-* `instance_type` - (Required) EC2 instance type for all instances in the instance group.
-* `bid_price` - (Optional) Bid price for each EC2 instance in the instance group, expressed in USD. By setting this attribute, the instance group is being declared as a Spot Instance, and will implicitly create a Spot request. Leave this blank to use On-Demand Instances.
-* `ebs_config` - (Optional) Configuration block(s) for EBS volumes attached to each instance in the instance group. Detailed below.
-* `instance_count` - (Optional) Target number of instances for the instance group. Must be 1 or 3. Defaults to 1. Launching with multiple master nodes is only supported in EMR version 5.23.0+, and requires this resource's `core_instance_group` to be configured. Public (Internet accessible) instances must be created in VPC subnets that have [map public IP on launch](/docs/providers/aws/r/subnet.html#map_public_ip_on_launch) enabled. Termination protection is automatically enabled when launched with multiple master nodes and Terraform must have the `termination_protection = false` configuration applied before destroying this resource.
-* `name` - (Optional) Friendly name given to the instance group.
-
-## ebs_config
-
-Attributes for the EBS volumes attached to each EC2 instance in the `master_instance_group` and `core_instance_group` configuration blocks:
+### Bootable Cluster
 
-* `size` - (Required) The volume size, in gibibytes (GiB).
-* `type` - (Required) The volume type. Valid options are `gp2`, `io1`, `standard` and `st1`. See [EBS Volume Types](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSVolumeTypes.html).
-* `iops` - (Optional) The number of I/O operations per second (IOPS) that the volume supports
-* `volumes_per_instance` - (Optional) The number of EBS volumes with this configuration to attach to each EC2 instance in the instance group (default is 1)
-
-## bootstrap_action
-
-* `name` - (Required) Name of the bootstrap action
-* `path` - (Required) Location of the script to run during a bootstrap action. Can be either a location in Amazon S3 or on a local file system
-* `args` - (Optional) List of command line arguments to pass to the bootstrap action script
-
-## step
-
-This argument is processed in [attribute-as-blocks mode](https://www.terraform.io/docs/configuration/attr-as-blocks.html).
-
-Attributes for step configuration
-
-* `action_on_failure` - (Required) The action to take if the step fails. Valid values: `TERMINATE_JOB_FLOW`, `TERMINATE_CLUSTER`, `CANCEL_AND_WAIT`, and `CONTINUE`
-* `hadoop_jar_step` - (Required) The JAR file used for the step. Defined below.
-* `name` - (Required) The name of the step.
-
-### hadoop_jar_step
-
-This argument is processed in [attribute-as-blocks mode](https://www.terraform.io/docs/configuration/attr-as-blocks.html).
-
-Attributes for Hadoop job step configuration
-
-* `args` - (Optional) List of command line arguments passed to the JAR file's main function when executed.
-* `jar` - (Required) Path to a JAR file run during the step.
-* `main_class` - (Optional) Name of the main class in the specified Java file. If not specified, the JAR file should specify a Main-Class in its manifest file.
-* `properties` - (Optional) Key-Value map of Java properties that are set when the step runs. You can use these properties to pass key value pairs to your main function.
-
-## master_instance_fleet Configuration Block
-
-Supported nested arguments for the `master_instance_fleet` configuration block:
-
-* `instance_type_configs` - (Optional) Configuration block for instance fleet
-* `launch_specifications` - (Optional) Configuration block for launch specification
-* `target_on_demand_capacity` - (Optional)  The target capacity of On-Demand units for the instance fleet, which determines how many On-Demand instances to provision.
-* `target_spot_capacity` - (Optional) The target capacity of Spot units for the instance fleet, which determines how many Spot instances to provision.
-* `name` - (Optional) Friendly name given to the instance fleet.
-
-## core_instance_fleet Configuration Block
-
-Supported nested arguments for the `core_instance_fleet` configuration block:
-
-* `instance_type_configs` - (Optional) Configuration block for instance fleet
-* `launch_specifications` - (Optional) Configuration block for launch specification
-* `target_on_demand_capacity` - (Optional)  The target capacity of On-Demand units for the instance fleet, which determines how many On-Demand instances to provision.
-* `target_spot_capacity` - (Optional) The target capacity of Spot units for the instance fleet, which determines how many Spot instances to provision.
-* `name` - (Optional) Friendly name given to the instance fleet.
-
-## instance_type_configs Configuration Block
-
-* `bid_price` - (Optional) The bid price for each EC2 Spot instance type as defined by `instance_type`. Expressed in USD. If neither `bid_price` nor `bid_price_as_percentage_of_on_demand_price` is provided, `bid_price_as_percentage_of_on_demand_price` defaults to 100%.
-* `bid_price_as_percentage_of_on_demand_price` - (Optional) The bid price, as a percentage of On-Demand price, for each EC2 Spot instance as defined by `instance_type`. Expressed as a number (for example, 20 specifies 20%). If neither `bid_price` nor `bid_price_as_percentage_of_on_demand_price` is provided, `bid_price_as_percentage_of_on_demand_price` defaults to 100%.
-* `configurations` - (Optional) A configuration classification that applies when provisioning cluster instances, which can include configurations for applications and software that run on the cluster. List of `configuration` blocks.
-* `ebs_config` - (Optional) Configuration block(s) for EBS volumes attached to each instance in the instance group. Detailed below.
-* `instance_type` - (Required) An EC2 instance type, such as m4.xlarge.
-* `weighted_capacity` - (Optional) The number of units that a provisioned instance of this type provides toward fulfilling the target capacities defined in `aws_emr_instance_fleet`.
-
-## configurations Configuration Block
-
-A configuration classification that applies when provisioning cluster instances, which can include configurations for applications and software that run on the cluster. See [Configuring Applications](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html).
-
-* `classification` - (Optional) The classification within a configuration.
-* `properties` - (Optional) A map of properties specified within a configuration classification
-
-## launch_specifications Configuration Block
-
-* `on_demand_specification` - (Optional) Configuration block for on demand instances launch specifications
-* `spot_specification` - (Optional) Configuration block for spot instances launch specifications
-
-## on_demand_specification  Configuration Block
-
-The launch specification for On-Demand instances in the instance fleet, which determines the allocation strategy.
-The instance fleet configuration is available only in Amazon EMR versions 4.8.0 and later, excluding 5.0.x versions. On-Demand instances allocation strategy is available in Amazon EMR version 5.12.1 and later.
-
-* `allocation_strategy` - (Required) Specifies the strategy to use in launching On-Demand instance fleets. Currently, the only option is `lowest-price` (the default), which launches the lowest price first.
-
-## spot_specification  Configuration Block
-
-The launch specification for Spot instances in the fleet, which determines the defined duration, provisioning timeout behavior, and allocation strategy.
-
-* `allocation_strategy` - (Required) Specifies the strategy to use in launching Spot instance fleets. Currently, the only option is `capacity-optimized` (the default), which launches instances from Spot instance pools with optimal capacity for the number of instances that are launching.
-* `block_duration_minutes` - (Optional) The defined duration for Spot instances (also known as Spot blocks) in minutes. When specified, the Spot instance does not terminate before the defined duration expires, and defined duration pricing for Spot instances applies. Valid values are 60, 120, 180, 240, 300, or 360. The duration period starts as soon as a Spot instance receives its instance ID. At the end of the duration, Amazon EC2 marks the Spot instance for termination and provides a Spot instance termination notice, which gives the instance a two-minute warning before it terminates.
-* `timeout_action` - (Required) The action to take when TargetSpotCapacity has not been fulfilled when the TimeoutDurationMinutes has expired; that is, when all Spot instances could not be provisioned within the Spot provisioning timeout. Valid values are `TERMINATE_CLUSTER` and `SWITCH_TO_ON_DEMAND`. SWITCH_TO_ON_DEMAND specifies that if no Spot instances are available, On-Demand Instances should be provisioned to fulfill any remaining Spot capacity.
-* `timeout_duration_minutes` - (Required) The spot provisioning timeout period in minutes. If Spot instances are not provisioned within this time period, the TimeOutAction is taken. Minimum value is 5 and maximum value is 1440. The timeout applies only during initial provisioning, when the cluster is first created.
-
-## Attributes Reference
-
-In addition to all arguments above, the following attributes are exported:
-
-* `arn`- The ARN of the cluster.
-* `id` - The ID of the EMR Cluster
-* `name` - The name of the cluster.
-* `release_label` - The release label for the Amazon EMR release.
-* `master_instance_group.0.id` - Master node type Instance Group ID, if using Instance Group for this node type.
-* `master_public_dns` - The public DNS name of the master EC2 instance.
-* `core_instance_group.0.id` - Core node type Instance Group ID, if using Instance Group for this node type.
-* `log_uri` - The path to the Amazon S3 location where logs for this cluster are stored.
-* `applications` - The applications installed on this cluster.
-* `ec2_attributes` - Provides information about the EC2 instances in a cluster grouped by category: key name, subnet ID, IAM instance profile, and so on.
-* `bootstrap_action` - A list of bootstrap actions that will be run before Hadoop is started on the cluster nodes.
-* `configurations` - The list of Configurations supplied to the EMR cluster.
-* `service_role` - The IAM role that will be assumed by the Amazon EMR service to access AWS resources on your behalf.
-* `visible_to_all_users` - Indicates whether the job flow is visible to all IAM users of the AWS account associated with the job flow.
-* `tags_all` - A map of tags assigned to the resource, including those inherited from the provider [`default_tags` configuration block](https://www.terraform.io/docs/providers/aws/index.html#default_tags-configuration-block).
-
-## Example bootable config
-
-**NOTE:** This configuration demonstrates a minimal configuration needed to
-boot an example EMR Cluster. It is not meant to display best practices. Please
-use at your own risk.
+**NOTE:** This configuration demonstrates a minimal configuration needed to boot an example EMR Cluster. It is not meant to display best practices. As with all examples, use at your own risk.
 
 ```terraform
 resource "aws_emr_cluster" "cluster" {
@@ -851,6 +607,225 @@ EOF
 }
 ```
 
+## Argument Reference
+
+The following arguments are required:
+
+* `name` - (Required) Name of the job flow.
+* `release_label` - (Required) Release label for the Amazon EMR release.
+* `service_role` - (Required) IAM role that will be assumed by the Amazon EMR service to access AWS resources.
+
+The following arguments are optional:
+
+* `additional_info` - (Optional) JSON string for selecting additional features such as adding proxy information. Note: Currently there is no API to retrieve the value of this argument after EMR cluster creation from provider, therefore Terraform cannot detect drift from the actual EMR cluster if its value is changed outside Terraform.
+* `applications` - (Optional) List of applications for the cluster. Valid values are: `Flink`, `Hadoop`, `Hive`, `Mahout`, `Pig`, `Spark`, and `JupyterHub` (as of EMR 5.14.0). Case insensitive.
+* `autoscaling_role` - (Optional) IAM role for automatic scaling policies. The IAM role provides permissions that the automatic scaling feature requires to launch and terminate EC2 instances in an instance group.
+* `bootstrap_action` - (Optional) Ordered list of bootstrap actions that will be run before Hadoop is started on the cluster nodes. See below.
+* `configurations` - (Optional) List of configurations supplied for the EMR cluster you are creating. Supply a configuration object for applications to override their default configuration. See [AWS Documentation](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html) for more information.
+* `configurations_json` - (Optional) JSON string for supplying list of configurations for the EMR cluster.
+
+~> **NOTE on `configurations_json`:** If the `Configurations` value is empty then you should skip the `Configurations` field instead of providing an empty list as a value, `"Configurations": []`.
+
+```terraform
+resource "aws_emr_cluster" "cluster" {
+  # ... other configuration ...
+
+  configurations_json = <<EOF
+  [
+    {
+      "Classification": "hadoop-env",
+      "Configurations": [
+        {
+          "Classification": "export",
+          "Properties": {
+            "JAVA_HOME": "/usr/lib/jvm/java-1.8.0"
+          }
+        }
+      ],
+      "Properties": {}
+    }
+  ]
+EOF
+}
+```
+
+* `core_instance_fleet` - (Optional) Configuration block to use an [Instance Fleet](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-fleet.html) for the core node type. Cannot be specified if any `core_instance_group` configuration blocks are set. Detailed below.
+* `core_instance_group` - (Optional) Configuration block to use an [Instance Group](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-group-configuration.html#emr-plan-instance-groups) for the [core node type](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-master-core-task-nodes.html#emr-plan-core).
+* `custom_ami_id` - (Optional) Custom Amazon Linux AMI for the cluster (instead of an EMR-owned AMI). Available in Amazon EMR version 5.7.0 and later.
+* `ebs_root_volume_size` - (Optional) Size in GiB of the EBS root device volume of the Linux AMI that is used for each EC2 instance. Available in Amazon EMR version 4.x and later.
+* `ec2_attributes` - (Optional) Attributes for the EC2 instances running the job flow. See below.
+* `keep_job_flow_alive_when_no_steps` - (Optional) Switch on/off run cluster with no steps or when all steps are complete (default is on)
+* `kerberos_attributes` - (Optional) Kerberos configuration for the cluster. See below.
+* `log_encryption_kms_key_id` - (Optional) AWS KMS customer master key (CMK) key ID or arn used for encrypting log files. This attribute is only available with EMR version 5.30.0 and later, excluding EMR 6.0.0.  
+* `log_uri` - (Optional) S3 bucket to write the log files of the job flow. If a value is not provided, logs are not created.
+* `master_instance_fleet` - (Optional) Configuration block to use an [Instance Fleet](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-fleet.html) for the master node type. Cannot be specified if any `master_instance_group` configuration blocks are set. Detailed below.
+* `master_instance_group` - (Optional) Configuration block to use an [Instance Group](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-instance-group-configuration.html#emr-plan-instance-groups) for the [master node type](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-master-core-task-nodes.html#emr-plan-master).
+* `scale_down_behavior` - (Optional) Way that individual Amazon EC2 instances terminate when an automatic scale-in activity occurs or an `instance group` is resized.
+* `security_configuration` - (Optional) Security configuration name to attach to the EMR cluster. Only valid for EMR clusters with `release_label` 4.8.0 or greater.
+* `step` - (Optional) List of steps to run when creating the cluster. See below. It is highly recommended to utilize the [lifecycle configuration block](https://www.terraform.io/docs/configuration/meta-arguments/lifecycle.html) with `ignore_changes` if other steps are being managed outside of Terraform. This argument is processed in [attribute-as-blocks mode](https://www.terraform.io/docs/configuration/attr-as-blocks.html).
+* `step_concurrency_level` - (Optional) Number of steps that can be executed concurrently. You can specify a maximum of 256 steps. Only valid for EMR clusters with `release_label` 5.28.0 or greater (default is 1).
+* `tags` - (Optional) list of tags to apply to the EMR Cluster. If configured with a provider [`default_tags` configuration block](https://www.terraform.io/docs/providers/aws/index.html#default_tags-configuration-block) present, tags with matching keys will overwrite those defined at the provider-level.
+* `termination_protection` - (Optional) Switch on/off termination protection (default is `false`, except when using multiple master nodes). Before attempting to destroy the resource when termination protection is enabled, this configuration must be applied with its value set to `false`.
+* `visible_to_all_users` - (Optional) Whether the job flow is visible to all IAM users of the AWS account associated with the job flow. Default value is `true`.
+
+### bootstrap_action
+
+* `args` - (Optional) List of command line arguments to pass to the bootstrap action script.
+* `name` - (Required) Name of the bootstrap action.
+* `path` - (Required) Location of the script to run during a bootstrap action. Can be either a location in Amazon S3 or on a local file system.
+
+### configurations
+
+A configuration classification that applies when provisioning cluster instances, which can include configurations for applications and software that run on the cluster. See [Configuring Applications](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html).
+
+* `classification` - (Optional) Classification within a configuration.
+* `properties` - (Optional) Map of properties specified within a configuration classification.
+
+### core_instance_fleet
+
+* `instance_type_configs` - (Optional) Configuration block for instance fleet.
+* `launch_specifications` - (Optional) Configuration block for launch specification.
+* `name` - (Optional) Friendly name given to the instance fleet.
+* `target_on_demand_capacity` - (Optional)  The target capacity of On-Demand units for the instance fleet, which determines how many On-Demand instances to provision.
+* `target_spot_capacity` - (Optional) Target capacity of Spot units for the instance fleet, which determines how many Spot instances to provision.
+
+#### instance_type_configs
+
+* `bid_price` - (Optional) Bid price for each EC2 Spot instance type as defined by `instance_type`. Expressed in USD. If neither `bid_price` nor `bid_price_as_percentage_of_on_demand_price` is provided, `bid_price_as_percentage_of_on_demand_price` defaults to 100%.
+* `bid_price_as_percentage_of_on_demand_price` - (Optional) Bid price, as a percentage of On-Demand price, for each EC2 Spot instance as defined by `instance_type`. Expressed as a number (for example, 20 specifies 20%). If neither `bid_price` nor `bid_price_as_percentage_of_on_demand_price` is provided, `bid_price_as_percentage_of_on_demand_price` defaults to 100%.
+* `configurations` - (Optional) Configuration classification that applies when provisioning cluster instances, which can include configurations for applications and software that run on the cluster. List of `configuration` blocks.
+* `ebs_config` - (Optional) Configuration block(s) for EBS volumes attached to each instance in the instance group. Detailed below.
+* `instance_type` - (Required) EC2 instance type, such as m4.xlarge.
+* `weighted_capacity` - (Optional) Number of units that a provisioned instance of this type provides toward fulfilling the target capacities defined in `aws_emr_instance_fleet`.
+
+#### launch_specifications
+
+* `on_demand_specification` - (Optional) Configuration block for on demand instances launch specifications.
+* `spot_specification` - (Optional) Configuration block for spot instances launch specifications.
+
+##### on_demand_specification
+
+The launch specification for On-Demand instances in the instance fleet, which determines the allocation strategy.
+The instance fleet configuration is available only in Amazon EMR versions 4.8.0 and later, excluding 5.0.x versions. On-Demand instances allocation strategy is available in Amazon EMR version 5.12.1 and later.
+
+* `allocation_strategy` - (Required) Specifies the strategy to use in launching On-Demand instance fleets. Currently, the only option is `lowest-price` (the default), which launches the lowest price first.
+
+##### spot_specification
+
+The launch specification for Spot instances in the fleet, which determines the defined duration, provisioning timeout behavior, and allocation strategy.
+
+* `allocation_strategy` - (Required) Specifies the strategy to use in launching Spot instance fleets. Currently, the only option is `capacity-optimized` (the default), which launches instances from Spot instance pools with optimal capacity for the number of instances that are launching.
+* `block_duration_minutes` - (Optional) Defined duration for Spot instances (also known as Spot blocks) in minutes. When specified, the Spot instance does not terminate before the defined duration expires, and defined duration pricing for Spot instances applies. Valid values are 60, 120, 180, 240, 300, or 360. The duration period starts as soon as a Spot instance receives its instance ID. At the end of the duration, Amazon EC2 marks the Spot instance for termination and provides a Spot instance termination notice, which gives the instance a two-minute warning before it terminates.
+* `timeout_action` - (Required) Action to take when TargetSpotCapacity has not been fulfilled when the TimeoutDurationMinutes has expired; that is, when all Spot instances could not be provisioned within the Spot provisioning timeout. Valid values are `TERMINATE_CLUSTER` and `SWITCH_TO_ON_DEMAND`. SWITCH_TO_ON_DEMAND specifies that if no Spot instances are available, On-Demand Instances should be provisioned to fulfill any remaining Spot capacity.
+* `timeout_duration_minutes` - (Required) Spot provisioning timeout period in minutes. If Spot instances are not provisioned within this time period, the TimeOutAction is taken. Minimum value is 5 and maximum value is 1440. The timeout applies only during initial provisioning, when the cluster is first created.
+
+### core_instance_group
+
+* `autoscaling_policy` - (Optional) String containing the [EMR Auto Scaling Policy](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-automatic-scaling.html) JSON.
+* `bid_price` - (Optional) Bid price for each EC2 instance in the instance group, expressed in USD. By setting this attribute, the instance group is being declared as a Spot Instance, and will implicitly create a Spot request. Leave this blank to use On-Demand Instances.
+* `ebs_config` - (Optional) Configuration block(s) for EBS volumes attached to each instance in the instance group. Detailed below.
+* `instance_count` - (Optional) Target number of instances for the instance group. Must be at least 1. Defaults to 1.
+* `instance_type` - (Required) EC2 instance type for all instances in the instance group.
+* `name` - (Optional) Friendly name given to the instance group.
+
+#### ebs_config
+
+* `iops` - (Optional) Number of I/O operations per second (IOPS) that the volume supports.
+* `size` - (Required) Volume size, in gibibytes (GiB).
+* `type` - (Required) Volume type. Valid options are `gp2`, `io1`, `standard` and `st1`. See [EBS Volume Types](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/EBSVolumeTypes.html).
+* `volumes_per_instance` - (Optional) Number of EBS volumes with this configuration to attach to each EC2 instance in the instance group (default is 1).
+
+### ec2_attributes
+
+Attributes for the Amazon EC2 instances running the job flow:
+
+* `additional_master_security_groups` - (Optional) String containing a comma separated list of additional Amazon EC2 security group IDs for the master node.
+* `additional_slave_security_groups` - (Optional) String containing a comma separated list of additional Amazon EC2 security group IDs for the slave nodes as a comma separated string.
+* `emr_managed_master_security_group` - (Optional) Identifier of the Amazon EC2 EMR-Managed security group for the master node.
+* `emr_managed_slave_security_group` - (Optional) Identifier of the Amazon EC2 EMR-Managed security group for the slave nodes.
+* `instance_profile` - (Required) Instance Profile for EC2 instances of the cluster assume this role.
+* `key_name` - (Optional) Amazon EC2 key pair that can be used to ssh to the master node as the user called `hadoop`.
+* `service_access_security_group` - (Optional) Identifier of the Amazon EC2 service-access security group - required when the cluster runs on a private subnet.
+* `subnet_id` - (Optional) VPC subnet id where you want the job flow to launch. Cannot specify the `cc1.4xlarge` instance type for nodes of a job flow launched in an Amazon VPC.
+* `subnet_ids` - (Optional) List of VPC subnet id-s where you want the job flow to launch.  Amazon EMR identifies the best Availability Zone to launch instances according to your fleet specifications.
+
+~> **NOTE on EMR-Managed security groups:** These security groups will have any missing inbound or outbound access rules added and maintained by AWS, to ensure proper communication between instances in a cluster. The EMR service will maintain these rules for groups provided in `emr_managed_master_security_group` and `emr_managed_slave_security_group`; attempts to remove the required rules may succeed, only for the EMR service to re-add them in a matter of minutes. This may cause Terraform to fail to destroy an environment that contains an EMR cluster, because the EMR service does not revoke rules added on deletion, leaving a cyclic dependency between the security groups that prevents their deletion. To avoid this, use the `revoke_rules_on_delete` optional attribute for any Security Group used in `emr_managed_master_security_group` and `emr_managed_slave_security_group`. See [Amazon EMR-Managed Security Groups](http://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-man-sec-groups.html) for more information about the EMR-managed security group rules.
+
+### kerberos_attributes
+
+* `ad_domain_join_password` - (Optional) Active Directory password for `ad_domain_join_user`. Terraform cannot perform drift detection of this configuration.
+* `ad_domain_join_user` - (Optional) Required only when establishing a cross-realm trust with an Active Directory domain. A user with sufficient privileges to join resources to the domain. Terraform cannot perform drift detection of this configuration.
+* `cross_realm_trust_principal_password` - (Optional) Required only when establishing a cross-realm trust with a KDC in a different realm. The cross-realm principal password, which must be identical across realms. Terraform cannot perform drift detection of this configuration.
+* `kdc_admin_password` - (Required) Password used within the cluster for the kadmin service on the cluster-dedicated KDC, which maintains Kerberos principals, password policies, and keytabs for the cluster. Terraform cannot perform drift detection of this configuration.
+* `realm` - (Required) Name of the Kerberos realm to which all nodes in a cluster belong. For example, `EC2.INTERNAL`
+
+### master_instance_fleet
+
+* `instance_type_configs` - (Optional) Configuration block for instance fleet.
+* `launch_specifications` - (Optional) Configuration block for launch specification.
+* `name` - (Optional) Friendly name given to the instance fleet.
+* `target_on_demand_capacity` - (Optional) Target capacity of On-Demand units for the instance fleet, which determines how many On-Demand instances to provision.
+* `target_spot_capacity` - (Optional) Target capacity of Spot units for the instance fleet, which determines how many Spot instances to provision.
+
+#### instance_type_configs
+
+See `instance_type_configs` above, under `core_instance_fleet`.
+
+#### launch_specifications
+
+See `launch_specifications` above, under `core_instance_fleet`.
+
+### master_instance_group
+
+Supported nested arguments for the `master_instance_group` configuration block:
+
+* `bid_price` - (Optional) Bid price for each EC2 instance in the instance group, expressed in USD. By setting this attribute, the instance group is being declared as a Spot Instance, and will implicitly create a Spot request. Leave this blank to use On-Demand Instances.
+* `ebs_config` - (Optional) Configuration block(s) for EBS volumes attached to each instance in the instance group. Detailed below.
+* `instance_count` - (Optional) Target number of instances for the instance group. Must be 1 or 3. Defaults to 1. Launching with multiple master nodes is only supported in EMR version 5.23.0+, and requires this resource's `core_instance_group` to be configured. Public (Internet accessible) instances must be created in VPC subnets that have [map public IP on launch](/docs/providers/aws/r/subnet.html#map_public_ip_on_launch) enabled. Termination protection is automatically enabled when launched with multiple master nodes and Terraform must have the `termination_protection = false` configuration applied before destroying this resource.
+* `instance_type` - (Required) EC2 instance type for all instances in the instance group.
+* `name` - (Optional) Friendly name given to the instance group.
+
+#### ebs_config
+
+See `ebs_config` under `core_instance_group` above.
+
+### step
+
+This argument is processed in [attribute-as-blocks mode](https://www.terraform.io/docs/configuration/attr-as-blocks.html).
+
+* `action_on_failure` - (Required) Action to take if the step fails. Valid values: `TERMINATE_JOB_FLOW`, `TERMINATE_CLUSTER`, `CANCEL_AND_WAIT`, and `CONTINUE`
+* `hadoop_jar_step` - (Required) JAR file used for the step. See below.
+* `name` - (Required) Name of the step.
+
+#### hadoop_jar_step
+
+This argument is processed in [attribute-as-blocks mode](https://www.terraform.io/docs/configuration/attr-as-blocks.html).
+
+* `args` - (Optional) List of command line arguments passed to the JAR file's main function when executed.
+* `jar` - (Required) Path to a JAR file run during the step.
+* `main_class` - (Optional) Name of the main class in the specified Java file. If not specified, the JAR file should specify a Main-Class in its manifest file.
+* `properties` - (Optional) Key-Value map of Java properties that are set when the step runs. You can use these properties to pass key value pairs to your main function.
+
+## Attributes Reference
+
+In addition to all arguments above, the following attributes are exported:
+
+* `applications` - Applications installed on this cluster.
+* `arn`- ARN of the cluster.
+* `bootstrap_action` - List of bootstrap actions that will be run before Hadoop is started on the cluster nodes.
+* `configurations` - List of Configurations supplied to the EMR cluster.
+* `core_instance_group.0.id` - Core node type Instance Group ID, if using Instance Group for this node type.
+* `ec2_attributes` - Provides information about the EC2 instances in a cluster grouped by category: key name, subnet ID, IAM instance profile, and so on.
+* `id` - ID of the cluster.
+* `log_uri` - Path to the Amazon S3 location where logs for this cluster are stored.
+* `master_instance_group.0.id` - Master node type Instance Group ID, if using Instance Group for this node type.
+* `master_public_dns` - Public DNS name of the master EC2 instance.
+* `name` - Name of the cluster.
+* `release_label` - Release label for the Amazon EMR release.
+* `service_role` - IAM role that will be assumed by the Amazon EMR service to access AWS resources on your behalf.
+* `tags_all` - Map of tags assigned to the resource, including those inherited from the provider [`default_tags` configuration block](https://www.terraform.io/docs/providers/aws/index.html#default_tags-configuration-block).
+* `visible_to_all_users` - Indicates whether the job flow is visible to all IAM users of the AWS account associated with the job flow.
+
 ## Import
 
 EMR clusters can be imported using the `id`, e.g.