diff --git a/.changelog/36523.txt b/.changelog/36523.txt
new file mode 100644
index 00000000000..b6b27b4d4ba
--- /dev/null
+++ b/.changelog/36523.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_emr_cluster: Add `unhealthy_node_replacement` argument
+```
\ No newline at end of file
diff --git a/internal/service/emr/block_public_access_configuration.go b/internal/service/emr/block_public_access_configuration.go
index eaaf6918388..dd58c39bbf7 100644
--- a/internal/service/emr/block_public_access_configuration.go
+++ b/internal/service/emr/block_public_access_configuration.go
@@ -14,11 +14,12 @@ import (
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
 	"github.com/hashicorp/terraform-provider-aws/internal/create"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 	"github.com/hashicorp/terraform-provider-aws/names"
 )
 
-// @SDKResource("aws_emr_block_public_access_configuration")
-func ResourceBlockPublicAccessConfiguration() *schema.Resource {
+// @SDKResource("aws_emr_block_public_access_configuration", name="Block Public Access Configuration")
+func resourceBlockPublicAccessConfiguration() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceBlockPublicAccessConfigurationCreate,
 		ReadWithoutTimeout:   resourceBlockPublicAccessConfigurationRead,
@@ -94,7 +95,7 @@ func resourceBlockPublicAccessConfigurationRead(ctx context.Context, d *schema.R
 
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	out, err := FindBlockPublicAccessConfiguration(ctx, conn)
+	out, err := findBlockPublicAccessConfiguration(ctx, conn)
 
 	if err != nil {
 		return create.AppendDiagError(diags, names.EMR, create.ErrActionReading, ResNameBlockPublicAccessConfiguration, d.Id(), err)
@@ -128,6 +129,20 @@ func resourceBlockPublicAccessConfigurationDelete(ctx context.Context, d *schema
 	return diags
 }
 
+func findBlockPublicAccessConfiguration(ctx context.Context, conn *emr.EMR) (*emr.GetBlockPublicAccessConfigurationOutput, error) {
+	input := &emr.GetBlockPublicAccessConfigurationInput{}
+	output, err := conn.GetBlockPublicAccessConfigurationWithContext(ctx, input)
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.BlockPublicAccessConfiguration == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output, nil
+}
+
 func findDefaultBlockPublicAccessConfiguration() *emr.BlockPublicAccessConfiguration {
 	defaultPort := int64(defaultPermittedPublicSecurityGroupRulePort)
 	defaultPortPointer := &defaultPort
diff --git a/internal/service/emr/cluster.go b/internal/service/emr/cluster.go
index 564799974d5..67a2592c7d0 100644
--- a/internal/service/emr/cluster.go
+++ b/internal/service/emr/cluster.go
@@ -37,7 +37,7 @@ import (
 
 // @SDKResource("aws_emr_cluster", name="Cluster")
 // @Tags(identifierAttribute="id")
-func ResourceCluster() *schema.Resource {
+func resourceCluster() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceClusterCreate,
 		ReadWithoutTimeout:   resourceClusterRead,
@@ -50,150 +50,90 @@ func ResourceCluster() *schema.Resource {
 
 		CustomizeDiff: verify.SetTagsDiff,
 
-		Schema: map[string]*schema.Schema{
-			"additional_info": {
-				Type:             schema.TypeString,
-				Optional:         true,
-				ForceNew:         true,
-				ValidateFunc:     validation.StringIsJSON,
-				DiffSuppressFunc: verify.SuppressEquivalentJSONDiffs,
-				StateFunc: func(v interface{}) string {
-					json, _ := structure.NormalizeJsonString(v)
-					return json
-				},
-			},
-			"applications": {
-				Type:     schema.TypeSet,
-				Optional: true,
-				ForceNew: true,
-				Elem:     &schema.Schema{Type: schema.TypeString},
-				Set:      schema.HashString,
-			},
-			"arn": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
-			"auto_termination_policy": {
-				Type:     schema.TypeList,
-				MaxItems: 1,
-				Optional: true,
-				Elem: &schema.Resource{
+		SchemaFunc: func() map[string]*schema.Schema {
+			instanceFleetConfigSchema := func() *schema.Resource {
+				return &schema.Resource{
 					Schema: map[string]*schema.Schema{
-						"idle_timeout": {
-							Type:         schema.TypeInt,
-							Optional:     true,
-							ValidateFunc: validation.IntBetween(60, 604800),
-						},
-					},
-				},
-			},
-			"autoscaling_role": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Optional: true,
-			},
-			"bootstrap_action": {
-				Type:     schema.TypeList,
-				Optional: true,
-				ForceNew: true,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"args": {
-							Type:     schema.TypeList,
-							Optional: true,
-							ForceNew: true,
-							Elem:     &schema.Schema{Type: schema.TypeString},
-						},
-						"name": {
-							Type:     schema.TypeString,
-							Required: true,
-						},
-						"path": {
-							Type:     schema.TypeString,
-							Required: true,
-						},
-					},
-				},
-			},
-			"cluster_state": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
-			"configurations": {
-				Type:          schema.TypeString,
-				ForceNew:      true,
-				Optional:      true,
-				ConflictsWith: []string{"configurations_json"},
-			},
-			"configurations_json": {
-				Type:             schema.TypeString,
-				Optional:         true,
-				ForceNew:         true,
-				ValidateFunc:     validation.StringIsJSON,
-				DiffSuppressFunc: verify.SuppressEquivalentJSONDiffs,
-				StateFunc: func(v interface{}) string {
-					json, _ := structure.NormalizeJsonString(v)
-					return json
-				},
-				ConflictsWith: []string{"configurations"},
-			},
-			"core_instance_fleet": {
-				Type:          schema.TypeList,
-				Optional:      true,
-				ForceNew:      true,
-				Computed:      true,
-				MaxItems:      1,
-				Elem:          instanceFleetConfigSchema(),
-				ConflictsWith: []string{"core_instance_group", "master_instance_group"},
-			},
-			"core_instance_group": {
-				Type:     schema.TypeList,
-				Optional: true,
-				Computed: true,
-				ForceNew: true,
-				MaxItems: 1,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"autoscaling_policy": {
-							Type:             schema.TypeString,
-							Optional:         true,
-							DiffSuppressFunc: verify.SuppressEquivalentJSONDiffs,
-							ValidateFunc:     validation.StringIsJSON,
-						},
-						"bid_price": {
+						"id": {
 							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
+							Computed: true,
 						},
-						"ebs_config": {
+						"instance_type_configs": {
 							Type:     schema.TypeSet,
 							Optional: true,
-							Computed: true,
 							ForceNew: true,
 							Elem: &schema.Resource{
 								Schema: map[string]*schema.Schema{
-									"iops": {
-										Type:     schema.TypeInt,
+									"bid_price": {
+										Type:     schema.TypeString,
 										Optional: true,
 										ForceNew: true,
 									},
-									"size": {
-										Type:     schema.TypeInt,
-										Required: true,
+									"bid_price_as_percentage_of_on_demand_price": {
+										Type:     schema.TypeFloat,
+										Optional: true,
 										ForceNew: true,
+										Default:  100,
 									},
-									"throughput": {
-										Type:     schema.TypeInt,
+									"configurations": {
+										Type:     schema.TypeSet,
+										Optional: true,
+										ForceNew: true,
+										Elem: &schema.Resource{
+											Schema: map[string]*schema.Schema{
+												"classification": {
+													Type:     schema.TypeString,
+													Optional: true,
+													ForceNew: true,
+												},
+												"properties": {
+													Type:     schema.TypeMap,
+													Optional: true,
+													ForceNew: true,
+													Elem:     &schema.Schema{Type: schema.TypeString},
+												},
+											},
+										},
+									},
+									"ebs_config": {
+										Type:     schema.TypeSet,
 										Optional: true,
+										Computed: true,
 										ForceNew: true,
+										Elem: &schema.Resource{
+											Schema: map[string]*schema.Schema{
+												"iops": {
+													Type:     schema.TypeInt,
+													Optional: true,
+													ForceNew: true,
+												},
+												"size": {
+													Type:     schema.TypeInt,
+													Required: true,
+													ForceNew: true,
+												},
+												"type": {
+													Type:         schema.TypeString,
+													Required:     true,
+													ForceNew:     true,
+													ValidateFunc: validEBSVolumeType(),
+												},
+												"volumes_per_instance": {
+													Type:     schema.TypeInt,
+													Optional: true,
+													ForceNew: true,
+													Default:  1,
+												},
+											},
+										},
+										Set: resourceClusterEBSHashConfig,
 									},
-									"type": {
-										Type:         schema.TypeString,
-										Required:     true,
-										ForceNew:     true,
-										ValidateFunc: validEBSVolumeType(),
+									"instance_type": {
+										Type:     schema.TypeString,
+										Required: true,
+										ForceNew: true,
 									},
-									"volumes_per_instance": {
+									"weighted_capacity": {
 										Type:     schema.TypeInt,
 										Optional: true,
 										ForceNew: true,
@@ -201,566 +141,631 @@ func ResourceCluster() *schema.Resource {
 									},
 								},
 							},
-							Set: resourceClusterEBSHashConfig,
-						},
-						"id": {
-							Type:     schema.TypeString,
-							Computed: true,
-						},
-						"instance_count": {
-							Type:         schema.TypeInt,
-							Optional:     true,
-							Default:      1,
-							ValidateFunc: validation.IntAtLeast(1),
+							Set: resourceInstanceTypeHashConfig,
 						},
-						"instance_type": {
-							Type:     schema.TypeString,
-							Required: true,
-							ForceNew: true,
-						},
-						"name": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
-						},
-					},
-				},
-			},
-			"custom_ami_id": {
-				Type:         schema.TypeString,
-				ForceNew:     true,
-				Optional:     true,
-				ValidateFunc: validCustomAMIID,
-			},
-			"ebs_root_volume_size": {
-				Type:     schema.TypeInt,
-				ForceNew: true,
-				Optional: true,
-			},
-			"ec2_attributes": {
-				Type:     schema.TypeList,
-				MaxItems: 1,
-				Optional: true,
-				ForceNew: true,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"additional_master_security_groups": {
-							Type:     schema.TypeString,
+						"launch_specifications": {
+							Type:     schema.TypeList,
 							Optional: true,
 							ForceNew: true,
+							MaxItems: 1,
+							Elem: &schema.Resource{
+								Schema: map[string]*schema.Schema{
+									"on_demand_specification": {
+										Type:     schema.TypeList,
+										Optional: true,
+										ForceNew: true,
+										MinItems: 1,
+										Elem: &schema.Resource{
+											Schema: map[string]*schema.Schema{
+												"allocation_strategy": {
+													Type:         schema.TypeString,
+													Required:     true,
+													ForceNew:     true,
+													ValidateFunc: validation.StringInSlice(emr.OnDemandProvisioningAllocationStrategy_Values(), false),
+												},
+											},
+										},
+									},
+									"spot_specification": {
+										Type:     schema.TypeList,
+										Optional: true,
+										ForceNew: true,
+										MinItems: 1,
+										Elem: &schema.Resource{
+											Schema: map[string]*schema.Schema{
+												"allocation_strategy": {
+													Type:         schema.TypeString,
+													ForceNew:     true,
+													Required:     true,
+													ValidateFunc: validation.StringInSlice(emr.SpotProvisioningAllocationStrategy_Values(), false),
+												},
+												"block_duration_minutes": {
+													Type:     schema.TypeInt,
+													Optional: true,
+													ForceNew: true,
+													Default:  0,
+												},
+												"timeout_action": {
+													Type:         schema.TypeString,
+													Required:     true,
+													ForceNew:     true,
+													ValidateFunc: validation.StringInSlice(emr.SpotProvisioningTimeoutAction_Values(), false),
+												},
+												"timeout_duration_minutes": {
+													Type:     schema.TypeInt,
+													ForceNew: true,
+													Required: true,
+												},
+											},
+										},
+									},
+								},
+							},
 						},
-						"additional_slave_security_groups": {
+						"name": {
 							Type:     schema.TypeString,
 							Optional: true,
 							ForceNew: true,
 						},
-						"emr_managed_master_security_group": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
+						"provisioned_on_demand_capacity": {
+							Type:     schema.TypeInt,
 							Computed: true,
 						},
-						"emr_managed_slave_security_group": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
+						"provisioned_spot_capacity": {
+							Type:     schema.TypeInt,
 							Computed: true,
 						},
-						"instance_profile": {
-							Type:     schema.TypeString,
-							Required: true,
-							ForceNew: true,
-						},
-						"key_name": {
-							Type:     schema.TypeString,
+						"target_on_demand_capacity": {
+							Type:     schema.TypeInt,
 							Optional: true,
 							ForceNew: true,
+							Default:  0,
 						},
-						"service_access_security_group": {
-							Type:     schema.TypeString,
+						"target_spot_capacity": {
+							Type:     schema.TypeInt,
 							Optional: true,
 							ForceNew: true,
-							Computed: true,
-						},
-						"subnet_id": {
-							Type:          schema.TypeString,
-							Optional:      true,
-							Computed:      true,
-							ForceNew:      true,
-							ConflictsWith: []string{"ec2_attributes.0.subnet_ids"},
-						},
-						"subnet_ids": {
-							Type:          schema.TypeSet,
-							Optional:      true,
-							Computed:      true,
-							ForceNew:      true,
-							Elem:          &schema.Schema{Type: schema.TypeString},
-							Set:           schema.HashString,
-							ConflictsWith: []string{"ec2_attributes.0.subnet_id"},
+							Default:  0,
 						},
 					},
+				}
+			}
+
+			return map[string]*schema.Schema{
+				"additional_info": {
+					Type:             schema.TypeString,
+					Optional:         true,
+					ForceNew:         true,
+					ValidateFunc:     validation.StringIsJSON,
+					DiffSuppressFunc: verify.SuppressEquivalentJSONDiffs,
+					StateFunc: func(v interface{}) string {
+						json, _ := structure.NormalizeJsonString(v)
+						return json
+					},
 				},
-			},
-			"keep_job_flow_alive_when_no_steps": {
-				Type:     schema.TypeBool,
-				ForceNew: true,
-				Optional: true,
-				Computed: true,
-			},
-			"kerberos_attributes": {
-				Type:     schema.TypeList,
-				MaxItems: 1,
-				Optional: true,
-				ForceNew: true,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"ad_domain_join_password": {
-							Type:      schema.TypeString,
-							Optional:  true,
-							Sensitive: true,
-							ForceNew:  true,
-						},
-						"ad_domain_join_user": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
-						},
-						"cross_realm_trust_principal_password": {
-							Type:      schema.TypeString,
-							Optional:  true,
-							Sensitive: true,
-							ForceNew:  true,
-						},
-						"kdc_admin_password": {
-							Type:      schema.TypeString,
-							Required:  true,
-							Sensitive: true,
-							ForceNew:  true,
+				"applications": {
+					Type:     schema.TypeSet,
+					Optional: true,
+					ForceNew: true,
+					Elem:     &schema.Schema{Type: schema.TypeString},
+				},
+				"arn": {
+					Type:     schema.TypeString,
+					Computed: true,
+				},
+				"auto_termination_policy": {
+					Type:     schema.TypeList,
+					MaxItems: 1,
+					Optional: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"idle_timeout": {
+								Type:         schema.TypeInt,
+								Optional:     true,
+								ValidateFunc: validation.IntBetween(60, 604800),
+							},
 						},
-						"realm": {
-							Type:     schema.TypeString,
-							Required: true,
-							ForceNew: true,
+					},
+				},
+				"autoscaling_role": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Optional: true,
+				},
+				"bootstrap_action": {
+					Type:     schema.TypeList,
+					Optional: true,
+					ForceNew: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"args": {
+								Type:     schema.TypeList,
+								Optional: true,
+								ForceNew: true,
+								Elem:     &schema.Schema{Type: schema.TypeString},
+							},
+							"name": {
+								Type:     schema.TypeString,
+								Required: true,
+							},
+							"path": {
+								Type:     schema.TypeString,
+								Required: true,
+							},
 						},
 					},
 				},
-			},
-			"list_steps_states": {
-				Type:     schema.TypeSet,
-				Optional: true,
-				Elem: &schema.Schema{
-					Type:         schema.TypeString,
-					ValidateFunc: validation.StringInSlice(emr.StepState_Values(), false),
+				"cluster_state": {
+					Type:     schema.TypeString,
+					Computed: true,
 				},
-			},
-			"log_encryption_kms_key_id": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Optional: true,
-			},
-			"log_uri": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Optional: true,
-				DiffSuppressFunc: func(k, old, new string, d *schema.ResourceData) bool {
-					// EMR uses a proprietary filesystem called EMRFS
-					// and both s3n & s3 protocols are mapped to that FS
-					// so they're equvivalent in this context (confirmed by AWS support)
-					old = strings.Replace(old, "s3n://", "s3://", -1)
-					return old == new
+				"configurations": {
+					Type:          schema.TypeString,
+					ForceNew:      true,
+					Optional:      true,
+					ConflictsWith: []string{"configurations_json"},
 				},
-			},
-			"master_instance_fleet": {
-				Type:          schema.TypeList,
-				Optional:      true,
-				ForceNew:      true,
-				Computed:      true,
-				MaxItems:      1,
-				Elem:          instanceFleetConfigSchema(),
-				ConflictsWith: []string{"core_instance_group", "master_instance_group"},
-			},
-			"master_instance_group": {
-				Type:     schema.TypeList,
-				Optional: true,
-				Computed: true,
-				ForceNew: true,
-				MaxItems: 1,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"bid_price": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
-						},
-						"ebs_config": {
-							Type:     schema.TypeSet,
-							Optional: true,
-							Computed: true,
-							ForceNew: true,
-							Elem: &schema.Resource{
-								Schema: map[string]*schema.Schema{
-									"iops": {
-										Type:     schema.TypeInt,
-										Optional: true,
-										ForceNew: true,
-									},
-									"size": {
-										Type:     schema.TypeInt,
-										Required: true,
-										ForceNew: true,
-									},
-									"throughput": {
-										Type:     schema.TypeInt,
-										Optional: true,
-										ForceNew: true,
-									},
-									"type": {
-										Type:         schema.TypeString,
-										Required:     true,
-										ForceNew:     true,
-										ValidateFunc: validEBSVolumeType(),
-									},
-									"volumes_per_instance": {
-										Type:     schema.TypeInt,
-										Optional: true,
-										ForceNew: true,
-										Default:  1,
+				"configurations_json": {
+					Type:             schema.TypeString,
+					Optional:         true,
+					ForceNew:         true,
+					ValidateFunc:     validation.StringIsJSON,
+					DiffSuppressFunc: verify.SuppressEquivalentJSONDiffs,
+					StateFunc: func(v interface{}) string {
+						json, _ := structure.NormalizeJsonString(v)
+						return json
+					},
+					ConflictsWith: []string{"configurations"},
+				},
+				"core_instance_fleet": {
+					Type:          schema.TypeList,
+					Optional:      true,
+					ForceNew:      true,
+					Computed:      true,
+					MaxItems:      1,
+					Elem:          instanceFleetConfigSchema(),
+					ConflictsWith: []string{"core_instance_group", "master_instance_group"},
+				},
+				"core_instance_group": {
+					Type:     schema.TypeList,
+					Optional: true,
+					Computed: true,
+					ForceNew: true,
+					MaxItems: 1,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"autoscaling_policy": {
+								Type:             schema.TypeString,
+								Optional:         true,
+								DiffSuppressFunc: verify.SuppressEquivalentJSONDiffs,
+								ValidateFunc:     validation.StringIsJSON,
+							},
+							"bid_price": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+							},
+							"ebs_config": {
+								Type:     schema.TypeSet,
+								Optional: true,
+								Computed: true,
+								ForceNew: true,
+								Elem: &schema.Resource{
+									Schema: map[string]*schema.Schema{
+										"iops": {
+											Type:     schema.TypeInt,
+											Optional: true,
+											ForceNew: true,
+										},
+										"size": {
+											Type:     schema.TypeInt,
+											Required: true,
+											ForceNew: true,
+										},
+										"throughput": {
+											Type:     schema.TypeInt,
+											Optional: true,
+											ForceNew: true,
+										},
+										"type": {
+											Type:         schema.TypeString,
+											Required:     true,
+											ForceNew:     true,
+											ValidateFunc: validEBSVolumeType(),
+										},
+										"volumes_per_instance": {
+											Type:     schema.TypeInt,
+											Optional: true,
+											ForceNew: true,
+											Default:  1,
+										},
 									},
 								},
+								Set: resourceClusterEBSHashConfig,
+							},
+							"id": {
+								Type:     schema.TypeString,
+								Computed: true,
+							},
+							"instance_count": {
+								Type:         schema.TypeInt,
+								Optional:     true,
+								Default:      1,
+								ValidateFunc: validation.IntAtLeast(1),
+							},
+							"instance_type": {
+								Type:     schema.TypeString,
+								Required: true,
+								ForceNew: true,
+							},
+							"name": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
 							},
-							Set: resourceClusterEBSHashConfig,
-						},
-						"id": {
-							Type:     schema.TypeString,
-							Computed: true,
-						},
-						"instance_count": {
-							Type:         schema.TypeInt,
-							Optional:     true,
-							ForceNew:     true,
-							Default:      1,
-							ValidateFunc: validation.IntInSlice([]int{1, 3}),
-						},
-						"instance_type": {
-							Type:     schema.TypeString,
-							Required: true,
-							ForceNew: true,
-						},
-						"name": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
 						},
 					},
 				},
-			},
-			"master_public_dns": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
-			"name": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Required: true,
-			},
-			"placement_group_config": {
-				Type:       schema.TypeList,
-				ForceNew:   true,
-				Optional:   true,
-				ConfigMode: schema.SchemaConfigModeAttr,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"instance_role": {
-							Type:         schema.TypeString,
-							ForceNew:     true,
-							Required:     true,
-							ValidateFunc: validation.StringInSlice(emr.InstanceRoleType_Values(), false),
-						},
-						"placement_strategy": {
-							Type:         schema.TypeString,
-							ForceNew:     true,
-							Optional:     true,
-							Computed:     true,
-							ValidateFunc: validation.StringInSlice(emr.PlacementGroupStrategy_Values(), false),
+				"custom_ami_id": {
+					Type:         schema.TypeString,
+					ForceNew:     true,
+					Optional:     true,
+					ValidateFunc: validCustomAMIID,
+				},
+				"ebs_root_volume_size": {
+					Type:     schema.TypeInt,
+					ForceNew: true,
+					Optional: true,
+				},
+				"ec2_attributes": {
+					Type:     schema.TypeList,
+					MaxItems: 1,
+					Optional: true,
+					ForceNew: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"additional_master_security_groups": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+							},
+							"additional_slave_security_groups": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+							},
+							"emr_managed_master_security_group": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+								Computed: true,
+							},
+							"emr_managed_slave_security_group": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+								Computed: true,
+							},
+							"instance_profile": {
+								Type:     schema.TypeString,
+								Required: true,
+								ForceNew: true,
+							},
+							"key_name": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+							},
+							"service_access_security_group": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+								Computed: true,
+							},
+							"subnet_id": {
+								Type:          schema.TypeString,
+								Optional:      true,
+								Computed:      true,
+								ForceNew:      true,
+								ConflictsWith: []string{"ec2_attributes.0.subnet_ids"},
+							},
+							"subnet_ids": {
+								Type:          schema.TypeSet,
+								Optional:      true,
+								Computed:      true,
+								ForceNew:      true,
+								Elem:          &schema.Schema{Type: schema.TypeString},
+								ConflictsWith: []string{"ec2_attributes.0.subnet_id"},
+							},
 						},
 					},
 				},
-			},
-			"release_label": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Required: true,
-			},
-			"scale_down_behavior": {
-				Type:         schema.TypeString,
-				ForceNew:     true,
-				Optional:     true,
-				Computed:     true,
-				ValidateFunc: validation.StringInSlice(emr.ScaleDownBehavior_Values(), false),
-			},
-			"security_configuration": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Optional: true,
-			},
-			"service_role": {
-				Type:     schema.TypeString,
-				ForceNew: true,
-				Required: true,
-			},
-			"step": {
-				Type:       schema.TypeList,
-				Optional:   true,
-				Computed:   true,
-				ForceNew:   true,
-				ConfigMode: schema.SchemaConfigModeAttr,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"action_on_failure": {
-							Type:         schema.TypeString,
-							Required:     true,
-							ForceNew:     true,
-							ValidateFunc: validation.StringInSlice(emr.ActionOnFailure_Values(), false),
-						},
-						"hadoop_jar_step": {
-							Type:       schema.TypeList,
-							MaxItems:   1,
-							Required:   true,
-							ForceNew:   true,
-							ConfigMode: schema.SchemaConfigModeAttr,
-							Elem: &schema.Resource{
-								Schema: map[string]*schema.Schema{
-									"args": {
-										Type:     schema.TypeList,
-										Optional: true,
-										ForceNew: true,
-										Elem:     &schema.Schema{Type: schema.TypeString},
-									},
-									"jar": {
-										Type:     schema.TypeString,
-										Required: true,
-										ForceNew: true,
-									},
-									"main_class": {
-										Type:     schema.TypeString,
-										Optional: true,
-										ForceNew: true,
-									},
-									"properties": {
-										Type:     schema.TypeMap,
-										Optional: true,
-										ForceNew: true,
-										Elem:     &schema.Schema{Type: schema.TypeString},
-									},
-								},
+				"keep_job_flow_alive_when_no_steps": {
+					Type:     schema.TypeBool,
+					ForceNew: true,
+					Optional: true,
+					Computed: true,
+				},
+				"kerberos_attributes": {
+					Type:     schema.TypeList,
+					MaxItems: 1,
+					Optional: true,
+					ForceNew: true,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"ad_domain_join_password": {
+								Type:      schema.TypeString,
+								Optional:  true,
+								Sensitive: true,
+								ForceNew:  true,
+							},
+							"ad_domain_join_user": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
+							},
+							"cross_realm_trust_principal_password": {
+								Type:      schema.TypeString,
+								Optional:  true,
+								Sensitive: true,
+								ForceNew:  true,
+							},
+							"kdc_admin_password": {
+								Type:      schema.TypeString,
+								Required:  true,
+								Sensitive: true,
+								ForceNew:  true,
+							},
+							"realm": {
+								Type:     schema.TypeString,
+								Required: true,
+								ForceNew: true,
 							},
-						},
-						"name": {
-							Type:     schema.TypeString,
-							Required: true,
-							ForceNew: true,
 						},
 					},
 				},
-			},
-			"step_concurrency_level": {
-				Type:         schema.TypeInt,
-				Optional:     true,
-				Default:      1,
-				ValidateFunc: validation.IntBetween(1, 256),
-			},
-			names.AttrTags:    tftags.TagsSchema(),
-			names.AttrTagsAll: tftags.TagsSchemaComputed(),
-			"termination_protection": {
-				Type:     schema.TypeBool,
-				Optional: true,
-				Computed: true,
-			},
-			"visible_to_all_users": {
-				Type:     schema.TypeBool,
-				Optional: true,
-				Default:  true,
-			},
-		},
-	}
-}
-
-func instanceFleetConfigSchema() *schema.Resource {
-	return &schema.Resource{
-		Schema: map[string]*schema.Schema{
-			"id": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
-			"instance_type_configs": {
-				Type:     schema.TypeSet,
-				Optional: true,
-				ForceNew: true,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"bid_price": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ForceNew: true,
-						},
-						"bid_price_as_percentage_of_on_demand_price": {
-							Type:     schema.TypeFloat,
-							Optional: true,
-							ForceNew: true,
-							Default:  100,
-						},
-						"configurations": {
-							Type:     schema.TypeSet,
-							Optional: true,
-							ForceNew: true,
-							Elem: &schema.Resource{
-								Schema: map[string]*schema.Schema{
-									"classification": {
-										Type:     schema.TypeString,
-										Optional: true,
-										ForceNew: true,
-									},
-									"properties": {
-										Type:     schema.TypeMap,
-										Optional: true,
-										ForceNew: true,
-										Elem:     &schema.Schema{Type: schema.TypeString},
-									},
-								},
+				"list_steps_states": {
+					Type:     schema.TypeSet,
+					Optional: true,
+					Elem: &schema.Schema{
+						Type:         schema.TypeString,
+						ValidateFunc: validation.StringInSlice(emr.StepState_Values(), false),
+					},
+				},
+				"log_encryption_kms_key_id": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Optional: true,
+				},
+				"log_uri": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Optional: true,
+					DiffSuppressFunc: func(k, old, new string, d *schema.ResourceData) bool {
+						// EMR uses a proprietary filesystem called EMRFS
+						// and both s3n & s3 protocols are mapped to that FS
+						// so they're equvivalent in this context (confirmed by AWS support)
+						old = strings.Replace(old, "s3n://", "s3://", -1)
+						return old == new
+					},
+				},
+				"master_instance_fleet": {
+					Type:          schema.TypeList,
+					Optional:      true,
+					ForceNew:      true,
+					Computed:      true,
+					MaxItems:      1,
+					Elem:          instanceFleetConfigSchema(),
+					ConflictsWith: []string{"core_instance_group", "master_instance_group"},
+				},
+				"master_instance_group": {
+					Type:     schema.TypeList,
+					Optional: true,
+					Computed: true,
+					ForceNew: true,
+					MaxItems: 1,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"bid_price": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
 							},
-						},
-						"ebs_config": {
-							Type:     schema.TypeSet,
-							Optional: true,
-							Computed: true,
-							ForceNew: true,
-							Elem: &schema.Resource{
-								Schema: map[string]*schema.Schema{
-									"iops": {
-										Type:     schema.TypeInt,
-										Optional: true,
-										ForceNew: true,
-									},
-									"size": {
-										Type:     schema.TypeInt,
-										Required: true,
-										ForceNew: true,
-									},
-									"type": {
-										Type:         schema.TypeString,
-										Required:     true,
-										ForceNew:     true,
-										ValidateFunc: validEBSVolumeType(),
-									},
-									"volumes_per_instance": {
-										Type:     schema.TypeInt,
-										Optional: true,
-										ForceNew: true,
-										Default:  1,
+							"ebs_config": {
+								Type:     schema.TypeSet,
+								Optional: true,
+								Computed: true,
+								ForceNew: true,
+								Elem: &schema.Resource{
+									Schema: map[string]*schema.Schema{
+										"iops": {
+											Type:     schema.TypeInt,
+											Optional: true,
+											ForceNew: true,
+										},
+										"size": {
+											Type:     schema.TypeInt,
+											Required: true,
+											ForceNew: true,
+										},
+										"throughput": {
+											Type:     schema.TypeInt,
+											Optional: true,
+											ForceNew: true,
+										},
+										"type": {
+											Type:         schema.TypeString,
+											Required:     true,
+											ForceNew:     true,
+											ValidateFunc: validEBSVolumeType(),
+										},
+										"volumes_per_instance": {
+											Type:     schema.TypeInt,
+											Optional: true,
+											ForceNew: true,
+											Default:  1,
+										},
 									},
 								},
+								Set: resourceClusterEBSHashConfig,
+							},
+							"id": {
+								Type:     schema.TypeString,
+								Computed: true,
+							},
+							"instance_count": {
+								Type:         schema.TypeInt,
+								Optional:     true,
+								ForceNew:     true,
+								Default:      1,
+								ValidateFunc: validation.IntInSlice([]int{1, 3}),
+							},
+							"instance_type": {
+								Type:     schema.TypeString,
+								Required: true,
+								ForceNew: true,
+							},
+							"name": {
+								Type:     schema.TypeString,
+								Optional: true,
+								ForceNew: true,
 							},
-							Set: resourceClusterEBSHashConfig,
-						},
-						"instance_type": {
-							Type:     schema.TypeString,
-							Required: true,
-							ForceNew: true,
-						},
-						"weighted_capacity": {
-							Type:     schema.TypeInt,
-							Optional: true,
-							ForceNew: true,
-							Default:  1,
 						},
 					},
 				},
-				Set: resourceInstanceTypeHashConfig,
-			},
-			"launch_specifications": {
-				Type:     schema.TypeList,
-				Optional: true,
-				ForceNew: true,
-				MaxItems: 1,
-				Elem: &schema.Resource{
-					Schema: map[string]*schema.Schema{
-						"on_demand_specification": {
-							Type:     schema.TypeList,
-							Optional: true,
-							ForceNew: true,
-							MinItems: 1,
-							Elem: &schema.Resource{
-								Schema: map[string]*schema.Schema{
-									"allocation_strategy": {
-										Type:         schema.TypeString,
-										Required:     true,
-										ForceNew:     true,
-										ValidateFunc: validation.StringInSlice(emr.OnDemandProvisioningAllocationStrategy_Values(), false),
-									},
-								},
+				"master_public_dns": {
+					Type:     schema.TypeString,
+					Computed: true,
+				},
+				"name": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Required: true,
+				},
+				"placement_group_config": {
+					Type:       schema.TypeList,
+					ForceNew:   true,
+					Optional:   true,
+					ConfigMode: schema.SchemaConfigModeAttr,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"instance_role": {
+								Type:         schema.TypeString,
+								ForceNew:     true,
+								Required:     true,
+								ValidateFunc: validation.StringInSlice(emr.InstanceRoleType_Values(), false),
+							},
+							"placement_strategy": {
+								Type:         schema.TypeString,
+								ForceNew:     true,
+								Optional:     true,
+								Computed:     true,
+								ValidateFunc: validation.StringInSlice(emr.PlacementGroupStrategy_Values(), false),
 							},
 						},
-						"spot_specification": {
-							Type:     schema.TypeList,
-							Optional: true,
-							ForceNew: true,
-							MinItems: 1,
-							Elem: &schema.Resource{
-								Schema: map[string]*schema.Schema{
-									"allocation_strategy": {
-										Type:         schema.TypeString,
-										ForceNew:     true,
-										Required:     true,
-										ValidateFunc: validation.StringInSlice(emr.SpotProvisioningAllocationStrategy_Values(), false),
-									},
-									"block_duration_minutes": {
-										Type:     schema.TypeInt,
-										Optional: true,
-										ForceNew: true,
-										Default:  0,
-									},
-									"timeout_action": {
-										Type:         schema.TypeString,
-										Required:     true,
-										ForceNew:     true,
-										ValidateFunc: validation.StringInSlice(emr.SpotProvisioningTimeoutAction_Values(), false),
-									},
-									"timeout_duration_minutes": {
-										Type:     schema.TypeInt,
-										ForceNew: true,
-										Required: true,
+					},
+				},
+				"release_label": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Required: true,
+				},
+				"scale_down_behavior": {
+					Type:         schema.TypeString,
+					ForceNew:     true,
+					Optional:     true,
+					Computed:     true,
+					ValidateFunc: validation.StringInSlice(emr.ScaleDownBehavior_Values(), false),
+				},
+				"security_configuration": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Optional: true,
+				},
+				"service_role": {
+					Type:     schema.TypeString,
+					ForceNew: true,
+					Required: true,
+				},
+				"step": {
+					Type:       schema.TypeList,
+					Optional:   true,
+					Computed:   true,
+					ForceNew:   true,
+					ConfigMode: schema.SchemaConfigModeAttr,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"action_on_failure": {
+								Type:         schema.TypeString,
+								Required:     true,
+								ForceNew:     true,
+								ValidateFunc: validation.StringInSlice(emr.ActionOnFailure_Values(), false),
+							},
+							"hadoop_jar_step": {
+								Type:       schema.TypeList,
+								MaxItems:   1,
+								Required:   true,
+								ForceNew:   true,
+								ConfigMode: schema.SchemaConfigModeAttr,
+								Elem: &schema.Resource{
+									Schema: map[string]*schema.Schema{
+										"args": {
+											Type:     schema.TypeList,
+											Optional: true,
+											ForceNew: true,
+											Elem:     &schema.Schema{Type: schema.TypeString},
+										},
+										"jar": {
+											Type:     schema.TypeString,
+											Required: true,
+											ForceNew: true,
+										},
+										"main_class": {
+											Type:     schema.TypeString,
+											Optional: true,
+											ForceNew: true,
+										},
+										"properties": {
+											Type:     schema.TypeMap,
+											Optional: true,
+											ForceNew: true,
+											Elem:     &schema.Schema{Type: schema.TypeString},
+										},
 									},
 								},
 							},
+							"name": {
+								Type:     schema.TypeString,
+								Required: true,
+								ForceNew: true,
+							},
 						},
 					},
 				},
-			},
-			"name": {
-				Type:     schema.TypeString,
-				Optional: true,
-				ForceNew: true,
-			},
-			"provisioned_on_demand_capacity": {
-				Type:     schema.TypeInt,
-				Computed: true,
-			},
-			"provisioned_spot_capacity": {
-				Type:     schema.TypeInt,
-				Computed: true,
-			},
-			"target_on_demand_capacity": {
-				Type:     schema.TypeInt,
-				Optional: true,
-				ForceNew: true,
-				Default:  0,
-			},
-			"target_spot_capacity": {
-				Type:     schema.TypeInt,
-				Optional: true,
-				ForceNew: true,
-				Default:  0,
-			},
+				"step_concurrency_level": {
+					Type:         schema.TypeInt,
+					Optional:     true,
+					Default:      1,
+					ValidateFunc: validation.IntBetween(1, 256),
+				},
+				names.AttrTags:    tftags.TagsSchema(),
+				names.AttrTagsAll: tftags.TagsSchemaComputed(),
+				"termination_protection": {
+					Type:     schema.TypeBool,
+					Optional: true,
+					Computed: true,
+				},
+				"unhealthy_node_replacement": {
+					Type:     schema.TypeBool,
+					Optional: true,
+					Default:  false,
+				},
+				"visible_to_all_users": {
+					Type:     schema.TypeBool,
+					Optional: true,
+					Default:  true,
+				},
+			}
 		},
 	}
 }
@@ -784,9 +789,16 @@ func resourceClusterCreate(ctx context.Context, d *schema.ResourceData, meta int
 	if v, ok := d.GetOk("termination_protection"); ok {
 		terminationProtection = v.(bool)
 	}
+
+	unhealthyNodeReplacement := false
+	if v, ok := d.GetOk("unhealthy_node_replacement"); ok {
+		unhealthyNodeReplacement = v.(bool)
+	}
+
 	instanceConfig := &emr.JobFlowInstancesConfig{
 		KeepJobFlowAliveWhenNoSteps: aws.Bool(keepJobFlowAliveWhenNoSteps),
 		TerminationProtected:        aws.Bool(terminationProtection),
+		UnhealthyNodeReplacement:    aws.Bool(unhealthyNodeReplacement),
 	}
 
 	if l := d.Get("master_instance_group").([]interface{}); len(l) > 0 && l[0] != nil {
@@ -1048,7 +1060,7 @@ func resourceClusterRead(ctx context.Context, d *schema.ResourceData, meta inter
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	cluster, err := FindClusterByID(ctx, conn, d.Id())
+	cluster, err := findClusterByID(ctx, conn, d.Id())
 
 	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] EMR Cluster (%s) not found, removing from state", d.Id())
@@ -1116,6 +1128,7 @@ func resourceClusterRead(ctx context.Context, d *schema.ResourceData, meta inter
 	d.Set("ebs_root_volume_size", cluster.EbsRootVolumeSize)
 	d.Set("scale_down_behavior", cluster.ScaleDownBehavior)
 	d.Set("termination_protection", cluster.TerminationProtected)
+	d.Set("unhealthy_node_replacement", cluster.UnhealthyNodeReplacement)
 	d.Set("step_concurrency_level", cluster.StepConcurrencyLevel)
 
 	d.Set("custom_ami_id", cluster.CustomAmiId)
@@ -1262,6 +1275,16 @@ func resourceClusterUpdate(ctx context.Context, d *schema.ResourceData, meta int
 		}
 	}
 
+	if d.HasChange("unhealthy_node_replacement") {
+		_, err := conn.SetUnhealthyNodeReplacementWithContext(ctx, &emr.SetUnhealthyNodeReplacementInput{
+			JobFlowIds:               []*string{aws.String(d.Id())},
+			UnhealthyNodeReplacement: aws.Bool(d.Get("unhealthy_node_replacement").(bool)),
+		})
+		if err != nil {
+			return sdkdiag.AppendErrorf(diags, "updating EMR Cluster (%s): setting unhealthy node replacement: %s", d.Id(), err)
+		}
+	}
+
 	if d.HasChange("core_instance_group.0.autoscaling_policy") {
 		autoscalingPolicyStr := d.Get("core_instance_group.0.autoscaling_policy").(string)
 		instanceGroupID := d.Get("core_instance_group.0.id").(string)
@@ -1439,6 +1462,127 @@ func resourceClusterDelete(ctx context.Context, d *schema.ResourceData, meta int
 	return diags
 }
 
+func findCluster(ctx context.Context, conn *emr.EMR, input *emr.DescribeClusterInput) (*emr.Cluster, error) {
+	output, err := conn.DescribeClusterWithContext(ctx, input)
+
+	if tfawserr.ErrCodeEquals(err, ErrCodeClusterNotFound) || tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "is not valid") {
+		return nil, &retry.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.Cluster == nil || output.Cluster.Status == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output.Cluster, nil
+}
+
+func findClusterByID(ctx context.Context, conn *emr.EMR, id string) (*emr.Cluster, error) {
+	input := &emr.DescribeClusterInput{
+		ClusterId: aws.String(id),
+	}
+
+	output, err := findCluster(ctx, conn, input)
+
+	if err != nil {
+		return nil, err
+	}
+
+	// Eventual consistency check.
+	if aws.StringValue(output.Id) != id {
+		return nil, &retry.NotFoundError{
+			LastRequest: input,
+		}
+	}
+
+	if state := aws.StringValue(output.Status.State); state == emr.ClusterStateTerminated || state == emr.ClusterStateTerminatedWithErrors {
+		return nil, &retry.NotFoundError{
+			Message:     state,
+			LastRequest: input,
+		}
+	}
+
+	return output, nil
+}
+
+func statusCluster(ctx context.Context, conn *emr.EMR, id string) retry.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		input := &emr.DescribeClusterInput{
+			ClusterId: aws.String(id),
+		}
+
+		output, err := findCluster(ctx, conn, input)
+
+		if tfresource.NotFound(err) {
+			return nil, "", nil
+		}
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		return output, aws.StringValue(output.Status.State), nil
+	}
+}
+
+func waitClusterCreated(ctx context.Context, conn *emr.EMR, id string) (*emr.Cluster, error) {
+	const (
+		timeout = 75 * time.Minute
+	)
+	stateConf := &retry.StateChangeConf{
+		Pending:    []string{emr.ClusterStateBootstrapping, emr.ClusterStateStarting},
+		Target:     []string{emr.ClusterStateRunning, emr.ClusterStateWaiting},
+		Refresh:    statusCluster(ctx, conn, id),
+		Timeout:    timeout,
+		MinTimeout: 10 * time.Second,
+		Delay:      30 * time.Second,
+	}
+
+	outputRaw, err := stateConf.WaitForStateContext(ctx)
+
+	if output, ok := outputRaw.(*emr.Cluster); ok {
+		if stateChangeReason := output.Status.StateChangeReason; stateChangeReason != nil {
+			tfresource.SetLastError(err, fmt.Errorf("%s: %s", aws.StringValue(stateChangeReason.Code), aws.StringValue(stateChangeReason.Message)))
+		}
+
+		return output, err
+	}
+
+	return nil, err
+}
+
+func waitClusterDeleted(ctx context.Context, conn *emr.EMR, id string) (*emr.Cluster, error) {
+	const (
+		timeout = 20 * time.Minute
+	)
+	stateConf := &retry.StateChangeConf{
+		Pending:    []string{emr.ClusterStateTerminating},
+		Target:     []string{emr.ClusterStateTerminated, emr.ClusterStateTerminatedWithErrors},
+		Refresh:    statusCluster(ctx, conn, id),
+		Timeout:    timeout,
+		MinTimeout: 10 * time.Second,
+		Delay:      30 * time.Second,
+	}
+
+	outputRaw, err := stateConf.WaitForStateContext(ctx)
+
+	if output, ok := outputRaw.(*emr.Cluster); ok {
+		if stateChangeReason := output.Status.StateChangeReason; stateChangeReason != nil {
+			tfresource.SetLastError(err, fmt.Errorf("%s: %s", aws.StringValue(stateChangeReason.Code), aws.StringValue(stateChangeReason.Message)))
+		}
+
+		return output, err
+	}
+
+	return nil, err
+}
+
 func expandApplications(apps []interface{}) []*emr.Application {
 	appOut := make([]*emr.Application, 0, len(apps))
 
diff --git a/internal/service/emr/cluster_test.go b/internal/service/emr/cluster_test.go
index 35befc7e5cc..8794aa57eff 100644
--- a/internal/service/emr/cluster_test.go
+++ b/internal/service/emr/cluster_test.go
@@ -1702,6 +1702,56 @@ func TestAccEMRCluster_InstanceFleetMaster_only(t *testing.T) {
 	})
 }
 
+func TestAccEMRCluster_unhealthyNodeReplacement(t *testing.T) {
+	ctx := acctest.Context(t)
+	var cluster emr.Cluster
+
+	resourceName := "aws_emr_cluster.test"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.EMRServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckClusterDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccClusterConfig_unhealthyNodeReplacement(rName, "true"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckClusterExists(ctx, resourceName, &cluster),
+					resource.TestCheckResourceAttr(resourceName, "unhealthy_node_replacement", "true"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"cluster_state", // Ignore RUNNING versus WAITING changes
+					"configurations",
+					"keep_job_flow_alive_when_no_steps",
+				},
+			},
+			{
+				Config: testAccClusterConfig_unhealthyNodeReplacement(rName, "false"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckClusterExists(ctx, resourceName, &cluster),
+					resource.TestCheckResourceAttr(resourceName, "unhealthy_node_replacement", "false"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"cluster_state", // Ignore RUNNING versus WAITING changes
+					"configurations",
+					"keep_job_flow_alive_when_no_steps",
+				},
+			},
+		},
+	})
+}
+
 func testAccCheckClusterDestroy(ctx context.Context) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		conn := acctest.Provider.Meta().(*conns.AWSClient).EMRConn(ctx)
@@ -4292,3 +4342,34 @@ resource "aws_emr_cluster" "test" {
 }
 `, rName))
 }
+
+func testAccClusterConfig_unhealthyNodeReplacement(rName, unhealthyNodeReplacement string) string {
+	return acctest.ConfigCompose(
+		testAccClusterConfig_baseVPC(rName, false),
+		fmt.Sprintf(`
+data "aws_partition" "current" {}
+
+resource "aws_emr_cluster" "test" {
+  applications                      = ["Spark"]
+  keep_job_flow_alive_when_no_steps = true
+  name                              = %[1]q
+  release_label                     = "emr-5.33.1"
+  service_role                      = "EMR_DefaultRole"
+  termination_protection            = false
+  unhealthy_node_replacement        = %[2]s
+
+  ec2_attributes {
+    instance_profile                  = "EMR_EC2_DefaultRole"
+    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
+  }
+
+  master_instance_group {
+    instance_type = "m4.large"
+  }
+
+  depends_on = [aws_route_table_association.test]
+}
+`, rName, unhealthyNodeReplacement))
+}
diff --git a/internal/service/emr/exports_test.go b/internal/service/emr/exports_test.go
new file mode 100644
index 00000000000..b5cec93e161
--- /dev/null
+++ b/internal/service/emr/exports_test.go
@@ -0,0 +1,24 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package emr
+
+// Exports for use in tests only.
+var (
+	ResourceBlockPublicAccessConfiguration = resourceBlockPublicAccessConfiguration
+	ResourceCluster                        = resourceCluster
+	ResourceInstanceFleet                  = resourceInstanceFleet
+	ResourceInstanceGroup                  = resourceInstanceGroup
+	ResourceManagedScalingPolicy           = resourceManagedScalingPolicy
+	ResourceSecurityConfiguration          = resourceSecurityConfiguration
+	ResourceStudio                         = resourceStudio
+	ResourceStudioSessionMapping           = resourceStudioSessionMapping
+
+	FetchInstanceGroup                 = fetchInstanceGroup
+	FindBlockPublicAccessConfiguration = findBlockPublicAccessConfiguration
+	FindClusterByID                    = findClusterByID
+	FindInstanceFleetByTwoPartKey      = findInstanceFleetByTwoPartKey
+	FindSecurityConfigurationByName    = findSecurityConfigurationByName
+	FindStudioByID                     = findStudioByID
+	FindStudioSessionMappingByIDOrName = findStudioSessionMappingByIDOrName
+)
diff --git a/internal/service/emr/find.go b/internal/service/emr/find.go
deleted file mode 100644
index 4c5fef4ffda..00000000000
--- a/internal/service/emr/find.go
+++ /dev/null
@@ -1,140 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: MPL-2.0
-
-package emr
-
-import (
-	"context"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/emr"
-	"github.com/hashicorp/aws-sdk-go-base/v2/awsv1shim/v2/tfawserr"
-	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
-	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
-)
-
-func FindCluster(ctx context.Context, conn *emr.EMR, input *emr.DescribeClusterInput) (*emr.Cluster, error) {
-	output, err := conn.DescribeClusterWithContext(ctx, input)
-
-	if tfawserr.ErrCodeEquals(err, ErrCodeClusterNotFound) || tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "is not valid") {
-		return nil, &retry.NotFoundError{
-			LastError:   err,
-			LastRequest: input,
-		}
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	if output == nil || output.Cluster == nil || output.Cluster.Status == nil {
-		return nil, tfresource.NewEmptyResultError(input)
-	}
-
-	return output.Cluster, nil
-}
-
-func FindClusterByID(ctx context.Context, conn *emr.EMR, id string) (*emr.Cluster, error) {
-	input := &emr.DescribeClusterInput{
-		ClusterId: aws.String(id),
-	}
-
-	output, err := FindCluster(ctx, conn, input)
-
-	if err != nil {
-		return nil, err
-	}
-
-	// Eventual consistency check.
-	if aws.StringValue(output.Id) != id {
-		return nil, &retry.NotFoundError{
-			LastRequest: input,
-		}
-	}
-
-	if state := aws.StringValue(output.Status.State); state == emr.ClusterStateTerminated || state == emr.ClusterStateTerminatedWithErrors {
-		return nil, &retry.NotFoundError{
-			Message:     state,
-			LastRequest: input,
-		}
-	}
-
-	return output, nil
-}
-
-func FindStudioByID(ctx context.Context, conn *emr.EMR, id string) (*emr.Studio, error) {
-	input := &emr.DescribeStudioInput{
-		StudioId: aws.String(id),
-	}
-
-	output, err := conn.DescribeStudioWithContext(ctx, input)
-
-	if tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "Studio does not exist") {
-		return nil, &retry.NotFoundError{
-			LastError:   err,
-			LastRequest: input,
-		}
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	if output == nil || output.Studio == nil {
-		return nil, tfresource.NewEmptyResultError(input)
-	}
-
-	return output.Studio, nil
-}
-
-func FindStudioSessionMappingByIDOrName(ctx context.Context, conn *emr.EMR, id string) (*emr.SessionMappingDetail, error) {
-	studioId, identityType, identityIdOrName, err := readStudioSessionMapping(id)
-	if err != nil {
-		return nil, err
-	}
-
-	input := &emr.GetStudioSessionMappingInput{
-		StudioId:     aws.String(studioId),
-		IdentityType: aws.String(identityType),
-	}
-
-	if isIdentityId(identityIdOrName) {
-		input.IdentityId = aws.String(identityIdOrName)
-	} else {
-		input.IdentityName = aws.String(identityIdOrName)
-	}
-
-	output, err := conn.GetStudioSessionMappingWithContext(ctx, input)
-
-	if tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "Studio session mapping does not exist") ||
-		tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "Studio does not exist") {
-		return nil, &retry.NotFoundError{
-			LastError:   err,
-			LastRequest: input,
-		}
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	if output == nil || output.SessionMapping == nil {
-		return nil, tfresource.NewEmptyResultError(input)
-	}
-
-	return output.SessionMapping, nil
-}
-
-func FindBlockPublicAccessConfiguration(ctx context.Context, conn *emr.EMR) (*emr.GetBlockPublicAccessConfigurationOutput, error) {
-	input := &emr.GetBlockPublicAccessConfigurationInput{}
-	output, err := conn.GetBlockPublicAccessConfigurationWithContext(ctx, input)
-	if err != nil {
-		return nil, err
-	}
-
-	if output == nil || output.BlockPublicAccessConfiguration == nil {
-		return nil, tfresource.NewEmptyResultError(input)
-	}
-
-	return output, nil
-}
diff --git a/internal/service/emr/instance_fleet.go b/internal/service/emr/instance_fleet.go
index 940f74a298b..66a5dca9f27 100644
--- a/internal/service/emr/instance_fleet.go
+++ b/internal/service/emr/instance_fleet.go
@@ -22,8 +22,8 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
-// @SDKResource("aws_emr_instance_fleet")
-func ResourceInstanceFleet() *schema.Resource {
+// @SDKResource("aws_emr_instance_fleet", name="Instance Fleet")
+func resourceInstanceFleet() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceInstanceFleetCreate,
 		ReadWithoutTimeout:   resourceInstanceFleetRead,
@@ -252,7 +252,7 @@ func resourceInstanceFleetRead(ctx context.Context, d *schema.ResourceData, meta
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	fleet, err := FindInstanceFleetByTwoPartKey(ctx, conn, d.Get("cluster_id").(string), d.Id())
+	fleet, err := findInstanceFleetByTwoPartKey(ctx, conn, d.Get("cluster_id").(string), d.Id())
 
 	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] EMR Instance Fleet (%s) not found, removing from state", d.Id())
@@ -347,7 +347,7 @@ func resourceInstanceFleetDelete(ctx context.Context, d *schema.ResourceData, me
 	return diags
 }
 
-func FindInstanceFleetByTwoPartKey(ctx context.Context, conn *emr.EMR, clusterID, fleetID string) (*emr.InstanceFleet, error) {
+func findInstanceFleetByTwoPartKey(ctx context.Context, conn *emr.EMR, clusterID, fleetID string) (*emr.InstanceFleet, error) {
 	input := &emr.ListInstanceFleetsInput{
 		ClusterId: aws.String(clusterID),
 	}
@@ -382,7 +382,7 @@ func FindInstanceFleetByTwoPartKey(ctx context.Context, conn *emr.EMR, clusterID
 
 func statusInstanceFleet(ctx context.Context, conn *emr.EMR, clusterID, fleetID string) retry.StateRefreshFunc {
 	return func() (interface{}, string, error) {
-		output, err := FindInstanceFleetByTwoPartKey(ctx, conn, clusterID, fleetID)
+		output, err := findInstanceFleetByTwoPartKey(ctx, conn, clusterID, fleetID)
 
 		if tfresource.NotFound(err) {
 			return nil, "", nil
diff --git a/internal/service/emr/instance_group.go b/internal/service/emr/instance_group.go
index bd156fcf0bf..83a147fe1a8 100644
--- a/internal/service/emr/instance_group.go
+++ b/internal/service/emr/instance_group.go
@@ -30,8 +30,8 @@ const (
 	instanceGroupUpdateTimeout = 30 * time.Minute
 )
 
-// @SDKResource("aws_emr_instance_group")
-func ResourceInstanceGroup() *schema.Resource {
+// @SDKResource("aws_emr_instance_group", name="Instance Group")
+func resourceInstanceGroup() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceInstanceGroupCreate,
 		ReadWithoutTimeout:   resourceInstanceGroupRead,
@@ -212,7 +212,7 @@ func resourceInstanceGroupRead(ctx context.Context, d *schema.ResourceData, meta
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	ig, err := FetchInstanceGroup(ctx, conn, d.Get("cluster_id").(string), d.Id())
+	ig, err := fetchInstanceGroup(ctx, conn, d.Get("cluster_id").(string), d.Id())
 
 	if tfresource.NotFound(err) {
 		log.Printf("[DEBUG] EMR Instance Group (%s) not found, removing", d.Id())
@@ -363,7 +363,7 @@ func resourceInstanceGroupDelete(ctx context.Context, d *schema.ResourceData, me
 
 func instanceGroupStateRefresh(ctx context.Context, conn *emr.EMR, clusterID, groupID string) retry.StateRefreshFunc {
 	return func() (interface{}, string, error) {
-		ig, err := FetchInstanceGroup(ctx, conn, clusterID, groupID)
+		ig, err := fetchInstanceGroup(ctx, conn, clusterID, groupID)
 		if err != nil {
 			return nil, "Not Found", err
 		}
@@ -377,7 +377,7 @@ func instanceGroupStateRefresh(ctx context.Context, conn *emr.EMR, clusterID, gr
 	}
 }
 
-func FetchInstanceGroup(ctx context.Context, conn *emr.EMR, clusterID, groupID string) (*emr.InstanceGroup, error) {
+func fetchInstanceGroup(ctx context.Context, conn *emr.EMR, clusterID, groupID string) (*emr.InstanceGroup, error) {
 	input := &emr.ListInstanceGroupsInput{ClusterId: aws.String(clusterID)}
 
 	var groups []*emr.InstanceGroup
diff --git a/internal/service/emr/managed_scaling_policy.go b/internal/service/emr/managed_scaling_policy.go
index 5dbd0fa73db..d0ea90d537a 100644
--- a/internal/service/emr/managed_scaling_policy.go
+++ b/internal/service/emr/managed_scaling_policy.go
@@ -17,12 +17,13 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/errs/sdkdiag"
 )
 
-// @SDKResource("aws_emr_managed_scaling_policy")
-func ResourceManagedScalingPolicy() *schema.Resource {
+// @SDKResource("aws_emr_managed_scaling_policy", name="Managed Scaling Policy")
+func resourceManagedScalingPolicy() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceManagedScalingPolicyCreate,
 		ReadWithoutTimeout:   resourceManagedScalingPolicyRead,
 		DeleteWithoutTimeout: resourceManagedScalingPolicyDelete,
+
 		Importer: &schema.ResourceImporter{
 			StateContext: schema.ImportStatePassthroughContext,
 		},
diff --git a/internal/service/emr/release_labels_data_source.go b/internal/service/emr/release_labels_data_source.go
index 443fc2ac82b..585145eca07 100644
--- a/internal/service/emr/release_labels_data_source.go
+++ b/internal/service/emr/release_labels_data_source.go
@@ -15,8 +15,8 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/errs/sdkdiag"
 )
 
-// @SDKDataSource("aws_emr_release_labels")
-func DataSourceReleaseLabels() *schema.Resource {
+// @SDKDataSource("aws_emr_release_labels", name="Release Labels")
+func dataSourceReleaseLabels() *schema.Resource {
 	return &schema.Resource{
 		ReadWithoutTimeout: dataSourceReleaseLabelsRead,
 
diff --git a/internal/service/emr/security_configuration.go b/internal/service/emr/security_configuration.go
index 3671fabd9d8..d3f4ecb3fe8 100644
--- a/internal/service/emr/security_configuration.go
+++ b/internal/service/emr/security_configuration.go
@@ -22,8 +22,8 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
-// @SDKResource("aws_emr_security_configuration")
-func ResourceSecurityConfiguration() *schema.Resource {
+// @SDKResource("aws_emr_security_configuration", name="Security Configuration")
+func resourceSecurityConfiguration() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceSecurityConfigurationCreate,
 		ReadWithoutTimeout:   resourceSecurityConfigurationRead,
@@ -93,7 +93,7 @@ func resourceSecurityConfigurationRead(ctx context.Context, d *schema.ResourceDa
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	output, err := FindSecurityConfigurationByName(ctx, conn, d.Id())
+	output, err := findSecurityConfigurationByName(ctx, conn, d.Id())
 
 	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] EMR Security Configuration (%s) not found, removing from state", d.Id())
@@ -133,7 +133,7 @@ func resourceSecurityConfigurationDelete(ctx context.Context, d *schema.Resource
 	return diags
 }
 
-func FindSecurityConfigurationByName(ctx context.Context, conn *emr.EMR, name string) (*emr.DescribeSecurityConfigurationOutput, error) {
+func findSecurityConfigurationByName(ctx context.Context, conn *emr.EMR, name string) (*emr.DescribeSecurityConfigurationOutput, error) {
 	input := &emr.DescribeSecurityConfigurationInput{
 		Name: aws.String(name),
 	}
diff --git a/internal/service/emr/service_package_gen.go b/internal/service/emr/service_package_gen.go
index 284c1409547..d2efe187acd 100644
--- a/internal/service/emr/service_package_gen.go
+++ b/internal/service/emr/service_package_gen.go
@@ -33,8 +33,9 @@ func (p *servicePackage) FrameworkResources(ctx context.Context) []*types.Servic
 func (p *servicePackage) SDKDataSources(ctx context.Context) []*types.ServicePackageSDKDataSource {
 	return []*types.ServicePackageSDKDataSource{
 		{
-			Factory:  DataSourceReleaseLabels,
+			Factory:  dataSourceReleaseLabels,
 			TypeName: "aws_emr_release_labels",
+			Name:     "Release Labels",
 		},
 	}
 }
@@ -42,11 +43,12 @@ func (p *servicePackage) SDKDataSources(ctx context.Context) []*types.ServicePac
 func (p *servicePackage) SDKResources(ctx context.Context) []*types.ServicePackageSDKResource {
 	return []*types.ServicePackageSDKResource{
 		{
-			Factory:  ResourceBlockPublicAccessConfiguration,
+			Factory:  resourceBlockPublicAccessConfiguration,
 			TypeName: "aws_emr_block_public_access_configuration",
+			Name:     "Block Public Access Configuration",
 		},
 		{
-			Factory:  ResourceCluster,
+			Factory:  resourceCluster,
 			TypeName: "aws_emr_cluster",
 			Name:     "Cluster",
 			Tags: &types.ServicePackageResourceTags{
@@ -54,23 +56,27 @@ func (p *servicePackage) SDKResources(ctx context.Context) []*types.ServicePacka
 			},
 		},
 		{
-			Factory:  ResourceInstanceFleet,
+			Factory:  resourceInstanceFleet,
 			TypeName: "aws_emr_instance_fleet",
+			Name:     "Instance Fleet",
 		},
 		{
-			Factory:  ResourceInstanceGroup,
+			Factory:  resourceInstanceGroup,
 			TypeName: "aws_emr_instance_group",
+			Name:     "Instance Group",
 		},
 		{
-			Factory:  ResourceManagedScalingPolicy,
+			Factory:  resourceManagedScalingPolicy,
 			TypeName: "aws_emr_managed_scaling_policy",
+			Name:     "Managed Scaling Policy",
 		},
 		{
-			Factory:  ResourceSecurityConfiguration,
+			Factory:  resourceSecurityConfiguration,
 			TypeName: "aws_emr_security_configuration",
+			Name:     "Security Configuration",
 		},
 		{
-			Factory:  ResourceStudio,
+			Factory:  resourceStudio,
 			TypeName: "aws_emr_studio",
 			Name:     "Studio",
 			Tags: &types.ServicePackageResourceTags{
@@ -78,8 +84,9 @@ func (p *servicePackage) SDKResources(ctx context.Context) []*types.ServicePacka
 			},
 		},
 		{
-			Factory:  ResourceStudioSessionMapping,
+			Factory:  resourceStudioSessionMapping,
 			TypeName: "aws_emr_studio_session_mapping",
+			Name:     "Studio Session Mapping",
 		},
 	}
 }
diff --git a/internal/service/emr/status.go b/internal/service/emr/status.go
deleted file mode 100644
index affeeace57d..00000000000
--- a/internal/service/emr/status.go
+++ /dev/null
@@ -1,33 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: MPL-2.0
-
-package emr
-
-import (
-	"context"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/emr"
-	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
-	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
-)
-
-func statusCluster(ctx context.Context, conn *emr.EMR, id string) retry.StateRefreshFunc {
-	return func() (interface{}, string, error) {
-		input := &emr.DescribeClusterInput{
-			ClusterId: aws.String(id),
-		}
-
-		output, err := FindCluster(ctx, conn, input)
-
-		if tfresource.NotFound(err) {
-			return nil, "", nil
-		}
-
-		if err != nil {
-			return nil, "", err
-		}
-
-		return output, aws.StringValue(output.Status.State), nil
-	}
-}
diff --git a/internal/service/emr/studio.go b/internal/service/emr/studio.go
index a0b8060f439..d2e07798ffe 100644
--- a/internal/service/emr/studio.go
+++ b/internal/service/emr/studio.go
@@ -25,12 +25,13 @@ import (
 
 // @SDKResource("aws_emr_studio", name="Studio")
 // @Tags(identifierAttribute="id")
-func ResourceStudio() *schema.Resource {
+func resourceStudio() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceStudioCreate,
 		ReadWithoutTimeout:   resourceStudioRead,
 		UpdateWithoutTimeout: resourceStudioUpdate,
 		DeleteWithoutTimeout: resourceStudioDelete,
+
 		Importer: &schema.ResourceImporter{
 			StateContext: schema.ImportStatePassthroughContext,
 		},
@@ -214,7 +215,8 @@ func resourceStudioRead(ctx context.Context, d *schema.ResourceData, meta interf
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	studio, err := FindStudioByID(ctx, conn, d.Id())
+	studio, err := findStudioByID(ctx, conn, d.Id())
+
 	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] EMR Studio (%s) not found, removing from state", d.Id())
 		d.SetId("")
@@ -265,3 +267,28 @@ func resourceStudioDelete(ctx context.Context, d *schema.ResourceData, meta inte
 
 	return diags
 }
+
+func findStudioByID(ctx context.Context, conn *emr.EMR, id string) (*emr.Studio, error) {
+	input := &emr.DescribeStudioInput{
+		StudioId: aws.String(id),
+	}
+
+	output, err := conn.DescribeStudioWithContext(ctx, input)
+
+	if tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "Studio does not exist") {
+		return nil, &retry.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.Studio == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output.Studio, nil
+}
diff --git a/internal/service/emr/studio_session_mapping.go b/internal/service/emr/studio_session_mapping.go
index e7eb05a39f4..f4aab2a7c2f 100644
--- a/internal/service/emr/studio_session_mapping.go
+++ b/internal/service/emr/studio_session_mapping.go
@@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/emr"
 	"github.com/hashicorp/aws-sdk-go-base/v2/awsv1shim/v2/tfawserr"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
@@ -20,13 +21,14 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/verify"
 )
 
-// @SDKResource("aws_emr_studio_session_mapping")
-func ResourceStudioSessionMapping() *schema.Resource {
+// @SDKResource("aws_emr_studio_session_mapping", name="Studio Session Mapping")
+func resourceStudioSessionMapping() *schema.Resource {
 	return &schema.Resource{
 		CreateWithoutTimeout: resourceStudioSessionMappingCreate,
 		ReadWithoutTimeout:   resourceStudioSessionMappingRead,
 		UpdateWithoutTimeout: resourceStudioSessionMappingUpdate,
 		DeleteWithoutTimeout: resourceStudioSessionMappingDelete,
+
 		Importer: &schema.ResourceImporter{
 			StateContext: schema.ImportStatePassthroughContext,
 		},
@@ -132,7 +134,8 @@ func resourceStudioSessionMappingRead(ctx context.Context, d *schema.ResourceDat
 	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).EMRConn(ctx)
 
-	mapping, err := FindStudioSessionMappingByIDOrName(ctx, conn, d.Id())
+	mapping, err := findStudioSessionMappingByIDOrName(ctx, conn, d.Id())
+
 	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] EMR Studio Session Mapping (%s) not found, removing from state", d.Id())
 		d.SetId("")
@@ -184,3 +187,41 @@ func resourceStudioSessionMappingDelete(ctx context.Context, d *schema.ResourceD
 
 	return diags
 }
+
+func findStudioSessionMappingByIDOrName(ctx context.Context, conn *emr.EMR, id string) (*emr.SessionMappingDetail, error) {
+	studioId, identityType, identityIdOrName, err := readStudioSessionMapping(id)
+	if err != nil {
+		return nil, err
+	}
+
+	input := &emr.GetStudioSessionMappingInput{
+		StudioId:     aws.String(studioId),
+		IdentityType: aws.String(identityType),
+	}
+
+	if isIdentityId(identityIdOrName) {
+		input.IdentityId = aws.String(identityIdOrName)
+	} else {
+		input.IdentityName = aws.String(identityIdOrName)
+	}
+
+	output, err := conn.GetStudioSessionMappingWithContext(ctx, input)
+
+	if tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "Studio session mapping does not exist") ||
+		tfawserr.ErrMessageContains(err, emr.ErrCodeInvalidRequestException, "Studio does not exist") {
+		return nil, &retry.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.SessionMapping == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output.SessionMapping, nil
+}
diff --git a/internal/service/emr/sweep.go b/internal/service/emr/sweep.go
index fe2f7c2758f..9085fd4e8f3 100644
--- a/internal/service/emr/sweep.go
+++ b/internal/service/emr/sweep.go
@@ -56,7 +56,7 @@ func sweepClusters(region string) error {
 				log.Printf("[ERROR] unsetting EMR Cluster (%s) termination protection: %s", id, err)
 			}
 
-			r := ResourceCluster()
+			r := resourceCluster()
 			d := r.Data(nil)
 			d.SetId(id)
 
@@ -103,7 +103,7 @@ func sweepStudios(region string) error {
 		}
 
 		for _, studio := range page.Studios {
-			r := ResourceStudio()
+			r := resourceStudio()
 			d := r.Data(nil)
 			d.SetId(aws.StringValue(studio.StudioId))
 
diff --git a/internal/service/emr/wait.go b/internal/service/emr/wait.go
deleted file mode 100644
index 84e7c0f6e65..00000000000
--- a/internal/service/emr/wait.go
+++ /dev/null
@@ -1,71 +0,0 @@
-// Copyright (c) HashiCorp, Inc.
-// SPDX-License-Identifier: MPL-2.0
-
-package emr
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/emr"
-	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
-	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
-)
-
-const (
-	ClusterCreatedTimeout    = 75 * time.Minute
-	ClusterCreatedMinTimeout = 10 * time.Second
-	ClusterCreatedDelay      = 30 * time.Second
-
-	ClusterDeletedTimeout    = 20 * time.Minute
-	ClusterDeletedMinTimeout = 10 * time.Second
-	ClusterDeletedDelay      = 30 * time.Second
-)
-
-func waitClusterCreated(ctx context.Context, conn *emr.EMR, id string) (*emr.Cluster, error) {
-	stateConf := &retry.StateChangeConf{
-		Pending:    []string{emr.ClusterStateBootstrapping, emr.ClusterStateStarting},
-		Target:     []string{emr.ClusterStateRunning, emr.ClusterStateWaiting},
-		Refresh:    statusCluster(ctx, conn, id),
-		Timeout:    ClusterCreatedTimeout,
-		MinTimeout: ClusterCreatedMinTimeout,
-		Delay:      ClusterCreatedDelay,
-	}
-
-	outputRaw, err := stateConf.WaitForStateContext(ctx)
-
-	if output, ok := outputRaw.(*emr.Cluster); ok {
-		if stateChangeReason := output.Status.StateChangeReason; stateChangeReason != nil {
-			tfresource.SetLastError(err, fmt.Errorf("%s: %s", aws.StringValue(stateChangeReason.Code), aws.StringValue(stateChangeReason.Message)))
-		}
-
-		return output, err
-	}
-
-	return nil, err
-}
-
-func waitClusterDeleted(ctx context.Context, conn *emr.EMR, id string) (*emr.Cluster, error) {
-	stateConf := &retry.StateChangeConf{
-		Pending:    []string{emr.ClusterStateTerminating},
-		Target:     []string{emr.ClusterStateTerminated, emr.ClusterStateTerminatedWithErrors},
-		Refresh:    statusCluster(ctx, conn, id),
-		Timeout:    ClusterDeletedTimeout,
-		MinTimeout: ClusterDeletedMinTimeout,
-		Delay:      ClusterDeletedDelay,
-	}
-
-	outputRaw, err := stateConf.WaitForStateContext(ctx)
-
-	if output, ok := outputRaw.(*emr.Cluster); ok {
-		if stateChangeReason := output.Status.StateChangeReason; stateChangeReason != nil {
-			tfresource.SetLastError(err, fmt.Errorf("%s: %s", aws.StringValue(stateChangeReason.Code), aws.StringValue(stateChangeReason.Message)))
-		}
-
-		return output, err
-	}
-
-	return nil, err
-}
diff --git a/website/docs/r/emr_cluster.html.markdown b/website/docs/r/emr_cluster.html.markdown
index f1fc2024cda..6ad4ecead66 100644
--- a/website/docs/r/emr_cluster.html.markdown
+++ b/website/docs/r/emr_cluster.html.markdown
@@ -663,6 +663,7 @@ EOF
 * `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://registry.terraform.io/providers/hashicorp/aws/latest/docs#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`.
+* `unhealthy_node_replacement` - (Optional) Whether whether Amazon EMR should gracefully replace core nodes that have degraded within the cluster. Default value is `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