package alicloud

import (
	"fmt"
	"github.com/aliyun/alibaba-cloud-sdk-go/sdk/requests"
	"github.com/aliyun/alibaba-cloud-sdk-go/services/ecs"
	"reflect"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/PaesslerAG/jsonpath"
	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"

	"github.com/hashicorp/terraform-plugin-sdk/helper/validation"

	"github.com/aliyun/terraform-provider-alicloud/alicloud/connectivity"
	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
)

func resourceAlicloudEmrV2Cluster() *schema.Resource {
	return &schema.Resource{
		Create: resourceAlicloudEmrV2ClusterCreate,
		Read:   resourceAlicloudEmrV2ClusterRead,
		Update: resourceAlicloudEmrV2ClusterUpdate,
		Delete: resourceAlicloudEmrV2ClusterDelete,
		Importer: &schema.ResourceImporter{
			State: schema.ImportStatePassthrough,
		},

		Timeouts: &schema.ResourceTimeout{
			Create: schema.DefaultTimeout(10 * time.Minute),
			Delete: schema.DefaultTimeout(5 * time.Minute),
		},

		Schema: map[string]*schema.Schema{
			"resource_group_id": {
				Type:     schema.TypeString,
				Optional: true,
				Computed: true,
			},
			"payment_type": {
				Type:         schema.TypeString,
				Optional:     true,
				Computed:     true,
				ValidateFunc: validation.StringInSlice([]string{"PayAsYouGo", "Subscription"}, false),
			},
			"subscription_config": {
				Type:     schema.TypeSet,
				Optional: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"payment_duration_unit": {
							Type:         schema.TypeString,
							Required:     true,
							ValidateFunc: validation.StringInSlice([]string{"Month", "Year"}, false),
						},
						"payment_duration": {
							Type:         schema.TypeInt,
							Required:     true,
							ValidateFunc: validation.IntInSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 12, 24, 36, 48, 60}),
						},
						"auto_renew": {
							Type:     schema.TypeBool,
							Optional: true,
						},
						"auto_pay_order": {
							Type:     schema.TypeBool,
							Optional: true,
						},
						"auto_renew_duration_unit": {
							Type:         schema.TypeString,
							Optional:     true,
							ValidateFunc: validation.StringInSlice([]string{"Month", "Year"}, false),
						},
						"auto_renew_duration": {
							Type:         schema.TypeInt,
							Optional:     true,
							ValidateFunc: validation.IntInSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 12, 24, 36, 48, 60}),
						},
					},
				},
				MaxItems: 1,
			},
			"cluster_type": {
				Type:         schema.TypeString,
				Required:     true,
				ForceNew:     true,
				ValidateFunc: validation.StringInSlice([]string{"DATALAKE", "OLAP", "DATAFLOW", "DATASERVING", "CUSTOM"}, false),
			},
			"release_version": {
				Type:     schema.TypeString,
				Required: true,
				ForceNew: true,
			},
			"cluster_name": {
				Type:     schema.TypeString,
				Required: true,
			},
			"log_collect_strategy": {
				Type:     schema.TypeString,
				Optional: true,
				Computed: true,
			},
			"deletion_protection": {
				Type:     schema.TypeBool,
				Optional: true,
			},
			"deploy_mode": {
				Type:         schema.TypeString,
				Optional:     true,
				ForceNew:     true,
				Computed:     true,
				ValidateFunc: validation.StringInSlice([]string{"NORMAL", "HA"}, false),
			},
			"security_mode": {
				Type:         schema.TypeString,
				Optional:     true,
				Computed:     true,
				ValidateFunc: validation.StringInSlice([]string{"NORMAL", "KERBEROS"}, false),
			},
			"applications": {
				Type:     schema.TypeSet,
				Required: true,
				ForceNew: true,
				Elem:     &schema.Schema{Type: schema.TypeString},
				MinItems: 1,
			},
			"application_configs": {
				Type:     schema.TypeSet,
				Optional: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"application_name": {
							Type:     schema.TypeString,
							Required: true,
						},
						"config_file_name": {
							Type:     schema.TypeString,
							Required: true,
						},
						"config_item_key": {
							Type:     schema.TypeString,
							Required: true,
						},
						"config_item_value": {
							Type:     schema.TypeString,
							Required: true,
						},
						"config_scope": {
							Type:         schema.TypeString,
							Optional:     true,
							ValidateFunc: validation.StringInSlice([]string{"CLUSTER", "NODE_GROUP"}, false),
						},
						"config_description": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"node_group_name": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"node_group_id": {
							Type:     schema.TypeString,
							Optional: true,
						},
					},
				},
			},
			"node_attributes": {
				Type:     schema.TypeSet,
				Required: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"vpc_id": {
							Type:     schema.TypeString,
							Required: true,
							ForceNew: true,
						},
						"zone_id": {
							Type:     schema.TypeString,
							Required: true,
							ForceNew: true,
						},
						"security_group_id": {
							Type:     schema.TypeString,
							Required: true,
							ForceNew: true,
						},
						"ram_role": {
							Type:     schema.TypeString,
							Required: true,
							ForceNew: true,
						},
						"key_pair_name": {
							Type:     schema.TypeString,
							Required: true,
							ForceNew: true,
						},
						"data_disk_encrypted": {
							Type:     schema.TypeBool,
							Optional: true,
							Computed: true,
							ForceNew: true,
						},
						"data_disk_kms_key_id": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ForceNew:     true,
							ValidateFunc: validation.NoZeroValues,
						},
						"system_disk_encrypted": {
							Type:     schema.TypeBool,
							Optional: true,
							ForceNew: true,
						},
						"system_disk_kms_key_id": {
							Type:         schema.TypeString,
							Optional:     true,
							ForceNew:     true,
							ValidateFunc: validation.NoZeroValues,
						},
					},
				},
				ForceNew: true,
			},
			"node_groups": {
				Type:     schema.TypeList,
				Required: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"node_group_type": {
							Type:         schema.TypeString,
							Required:     true,
							ValidateFunc: StringInSlice([]string{"MASTER", "CORE", "TASK", "GATEWAY", "MASTER-EXTEND"}, false),
						},
						"node_group_name": {
							Type:     schema.TypeString,
							Required: true,
						},
						"payment_type": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ValidateFunc: validation.StringInSlice([]string{"PayAsYouGo", "Subscription"}, false),
						},
						"deployment_set_strategy": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ValidateFunc: validation.StringInSlice([]string{"NONE", "CLUSTER", "NODE_GROUP"}, false),
						},
						"node_resize_strategy": {
							Type:         schema.TypeString,
							Optional:     true,
							Computed:     true,
							ValidateFunc: validation.StringInSlice([]string{"PRIORITY", "COST_OPTIMIZED"}, false),
						},
						"subscription_config": {
							Type:     schema.TypeSet,
							Optional: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"payment_duration_unit": {
										Type:         schema.TypeString,
										Required:     true,
										ValidateFunc: validation.StringInSlice([]string{"Month", "Year"}, false),
									},
									"payment_duration": {
										Type:         schema.TypeInt,
										Required:     true,
										ValidateFunc: validation.IntInSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 12, 24, 36, 48, 60}),
									},
									"auto_renew": {
										Type:     schema.TypeBool,
										Optional: true,
									},
									"auto_pay_order": {
										Type:     schema.TypeBool,
										Optional: true,
									},
									"auto_renew_duration_unit": {
										Type:         schema.TypeString,
										Optional:     true,
										ValidateFunc: validation.StringInSlice([]string{"Month", "Year"}, false),
									},
									"auto_renew_duration": {
										Type:         schema.TypeInt,
										Optional:     true,
										ValidateFunc: validation.IntInSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 12, 24, 36, 48, 60}),
									},
								},
							},
							MaxItems: 1,
						},
						"private_pool_options": {
							Type:     schema.TypeSet,
							Optional: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"private_pool_ids": {
										Type:     schema.TypeSet,
										Optional: true,
										Elem:     &schema.Schema{Type: schema.TypeString},
									},
									"match_criteria": {
										Type:         schema.TypeString,
										Optional:     true,
										ValidateFunc: validation.StringInSlice([]string{"Open", "Target", "None"}, false),
									},
								},
							},
							MaxItems: 1,
						},
						"spot_strategy": {
							Type:         schema.TypeString,
							Optional:     true,
							ValidateFunc: StringInSlice([]string{"NoSpot", "SpotWithPriceLimit", "SpotAsPriceGo"}, false),
						},
						"spot_bid_prices": {
							Type:     schema.TypeSet,
							Optional: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"instance_type": {
										Type:     schema.TypeString,
										Required: true,
									},
									"bid_price": {
										Type:     schema.TypeInt,
										Required: true,
									},
								},
							},
						},
						"vswitch_ids": {
							Type:     schema.TypeSet,
							Optional: true,
							Elem:     &schema.Schema{Type: schema.TypeString},
						},
						"with_public_ip": {
							Type:     schema.TypeBool,
							Optional: true,
							Computed: true,
						},
						"additional_security_group_ids": {
							Type:     schema.TypeSet,
							Optional: true,
							Elem:     &schema.Schema{Type: schema.TypeString},
						},
						"instance_types": {
							Type:     schema.TypeSet,
							Required: true,
							Elem:     &schema.Schema{Type: schema.TypeString},
						},
						"node_count": {
							Type:     schema.TypeInt,
							Required: true,
						},
						"system_disk": {
							Type:     schema.TypeSet,
							Required: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"category": {
										Type:         schema.TypeString,
										Required:     true,
										ValidateFunc: validation.StringInSlice([]string{"cloud_essd", "cloud_efficiency", "cloud_ssd"}, false),
									},
									"size": {
										Type:     schema.TypeInt,
										Required: true,
									},
									"performance_level": {
										Type:         schema.TypeString,
										Optional:     true,
										Computed:     true,
										ValidateFunc: validation.StringInSlice([]string{"PL0", "PL1", "PL2", "PL3"}, false),
									},
									"count": {
										Type:     schema.TypeInt,
										Optional: true,
										Computed: true,
									},
								},
							},
							MaxItems: 1,
						},
						"data_disks": {
							Type:     schema.TypeSet,
							Required: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"category": {
										Type:         schema.TypeString,
										Required:     true,
										ValidateFunc: validation.StringInSlice([]string{"cloud_efficiency", "cloud_ssd", "cloud_essd", "cloud", "local_hdd_pro", "local_disk", "local_ssd_pro"}, false),
									},
									"size": {
										Type:     schema.TypeInt,
										Required: true,
									},
									"performance_level": {
										Type:         schema.TypeString,
										Optional:     true,
										Computed:     true,
										ValidateFunc: validation.StringInSlice([]string{"PL0", "PL1", "PL2", "PL3"}, false),
									},
									"count": {
										Type:     schema.TypeInt,
										Optional: true,
										Computed: true,
									},
								},
							},
						},
						"graceful_shutdown": {
							Type:     schema.TypeBool,
							Optional: true,
							Computed: true,
						},
						"spot_instance_remedy": {
							Type:     schema.TypeBool,
							Optional: true,
							Computed: true,
						},
						"cost_optimized_config": {
							Type:     schema.TypeSet,
							Optional: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"on_demand_base_capacity": {
										Type:     schema.TypeInt,
										Required: true,
									},
									"on_demand_percentage_above_base_capacity": {
										Type:     schema.TypeInt,
										Required: true,
									},
									"spot_instance_pools": {
										Type:     schema.TypeInt,
										Required: true,
									},
								},
							},
							MaxItems: 1,
						},
						"auto_scaling_policy": {
							Type:     schema.TypeList,
							Optional: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"scaling_rules": {
										Type:     schema.TypeList,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"rule_name": {
													Type:     schema.TypeString,
													Required: true,
												},
												"trigger_type": {
													Type:         schema.TypeString,
													Required:     true,
													ValidateFunc: StringInSlice([]string{"TIME_TRIGGER", "METRICS_TRIGGER"}, false),
												},
												"activity_type": {
													Type:         schema.TypeString,
													Required:     true,
													ValidateFunc: StringInSlice([]string{"SCALE_OUT", "SCALE_IN"}, false),
												},
												"adjustment_type": {
													Type:         schema.TypeString,
													Optional:     true,
													ValidateFunc: StringInSlice([]string{"CHANGE_IN_CAPACITY", "EXACT_CAPACITY"}, false),
												},
												"adjustment_value": {
													Type:         schema.TypeInt,
													Required:     true,
													ValidateFunc: IntBetween(1, 5000),
												},
												"min_adjustment_value": {
													Type:     schema.TypeInt,
													Optional: true,
												},
												"time_trigger": {
													Type:     schema.TypeList,
													Optional: true,
													Elem: &schema.Resource{
														Schema: map[string]*schema.Schema{
															"launch_time": {
																Type:     schema.TypeString,
																Required: true,
															},
															"start_time": {
																Type:     schema.TypeString,
																Optional: true,
															},
															"end_time": {
																Type:     schema.TypeString,
																Optional: true,
															},
															"launch_expiration_time": {
																Type:         schema.TypeInt,
																Optional:     true,
																ValidateFunc: IntBetween(0, 3600),
															},
															"recurrence_type": {
																Type:         schema.TypeString,
																Optional:     true,
																ValidateFunc: StringInSlice([]string{"MINUTELY", "HOURLY", "DAILY", "WEEKLY", "MONTHLY"}, false),
															},
															"recurrence_value": {
																Type:     schema.TypeString,
																Optional: true,
															},
														},
													},
													MaxItems: 1,
												},
												"metrics_trigger": {
													Type:     schema.TypeList,
													Optional: true,
													Elem: &schema.Resource{
														Schema: map[string]*schema.Schema{
															"time_window": {
																Type:         schema.TypeInt,
																Required:     true,
																ValidateFunc: IntBetween(30, 1800),
															},
															"evaluation_count": {
																Type:         schema.TypeInt,
																Required:     true,
																ValidateFunc: IntBetween(1, 5),
															},
															"cool_down_interval": {
																Type:         schema.TypeInt,
																Optional:     true,
																ValidateFunc: IntBetween(0, 10800),
															},
															"condition_logic_operator": {
																Type:         schema.TypeString,
																Optional:     true,
																ValidateFunc: StringInSlice([]string{"And", "Or"}, false),
															},
															"time_constraints": {
																Type:     schema.TypeList,
																Optional: true,
																Elem: &schema.Resource{
																	Schema: map[string]*schema.Schema{
																		"start_time": {
																			Type:     schema.TypeString,
																			Optional: true,
																		},
																		"end_time": {
																			Type:     schema.TypeString,
																			Optional: true,
																		},
																	},
																},
															},
															"conditions": {
																Type:     schema.TypeList,
																Optional: true,
																Elem: &schema.Resource{
																	Schema: map[string]*schema.Schema{
																		"metric_name": {
																			Type:     schema.TypeString,
																			Required: true,
																		},
																		"statistics": {
																			Type:         schema.TypeString,
																			Required:     true,
																			ValidateFunc: StringInSlice([]string{"MAX", "MIN", "AVG"}, false),
																		},
																		"comparison_operator": {
																			Type:         schema.TypeString,
																			Required:     true,
																			ValidateFunc: StringInSlice([]string{"EQ", "NE", "GT", "LT", "GE", "LE"}, false),
																		},
																		"threshold": {
																			Type:     schema.TypeFloat,
																			Required: true,
																		},
																		"tags": {
																			Type:     schema.TypeList,
																			Optional: true,
																			Elem: &schema.Resource{
																				Schema: map[string]*schema.Schema{
																					"key": {
																						Type:     schema.TypeString,
																						Required: true,
																					},
																					"value": {
																						Type:     schema.TypeString,
																						Optional: true,
																					},
																				},
																			},
																		},
																	},
																},
															},
														},
													},
													MaxItems: 1,
												},
											},
										},
									},
									"constraints": {
										Type:     schema.TypeList,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"max_capacity": {
													Type:         schema.TypeInt,
													Optional:     true,
													ValidateFunc: IntBetween(0, 2000),
												},
												"min_capacity": {
													Type:         schema.TypeInt,
													Optional:     true,
													ValidateFunc: IntBetween(0, 2000),
												},
											},
										},
										MaxItems: 1,
									},
								},
							},
							MaxItems: 1,
						},
						"ack_config": {
							Type:     schema.TypeList,
							Optional: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"ack_instance_id": {
										Type:     schema.TypeString,
										Required: true,
									},
									"node_selectors": {
										Type:     schema.TypeSet,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"key": {
													Type:     schema.TypeString,
													Required: true,
												},
												"value": {
													Type:     schema.TypeString,
													Optional: true,
												},
											},
										},
									},
									"tolerations": {
										Type:     schema.TypeList,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"key": {
													Type:     schema.TypeString,
													Optional: true,
												},
												"value": {
													Type:     schema.TypeString,
													Optional: true,
												},
												"operator": {
													Type:     schema.TypeString,
													Optional: true,
												},
												"effect": {
													Type:     schema.TypeString,
													Optional: true,
												},
											},
										},
									},
									"namespace": {
										Type:     schema.TypeString,
										Required: true,
									},
									"request_cpu": {
										Type:     schema.TypeFloat,
										Required: true,
									},
									"request_memory": {
										Type:     schema.TypeFloat,
										Required: true,
									},
									"limit_cpu": {
										Type:     schema.TypeFloat,
										Required: true,
									},
									"limit_memory": {
										Type:     schema.TypeFloat,
										Required: true,
									},
									"custom_labels": {
										Type:     schema.TypeSet,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"key": {
													Type:     schema.TypeString,
													Required: true,
												},
												"value": {
													Type:     schema.TypeString,
													Optional: true,
												},
											},
										},
									},
									"custom_annotations": {
										Type:     schema.TypeSet,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"key": {
													Type:     schema.TypeString,
													Required: true,
												},
												"value": {
													Type:     schema.TypeString,
													Optional: true,
												},
											},
										},
									},
									"pvcs": {
										Type:     schema.TypeList,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"name": {
													Type:     schema.TypeString,
													Required: true,
												},
												"path": {
													Type:     schema.TypeString,
													Required: true,
												},
												"data_disk_storage_class": {
													Type:     schema.TypeString,
													Required: true,
												},
												"data_disk_size": {
													Type:     schema.TypeInt,
													Required: true,
												},
											},
										},
									},
									"volumes": {
										Type:     schema.TypeList,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"name": {
													Type:     schema.TypeString,
													Required: true,
												},
												"path": {
													Type:     schema.TypeString,
													Required: true,
												},
												"type": {
													Type:     schema.TypeString,
													Required: true,
												},
											},
										},
									},
									"volume_mounts": {
										Type:     schema.TypeList,
										Optional: true,
										Elem: &schema.Resource{
											Schema: map[string]*schema.Schema{
												"name": {
													Type:     schema.TypeString,
													Required: true,
												},
												"path": {
													Type:     schema.TypeString,
													Required: true,
												},
											},
										},
									},
									"pre_start_command": {
										Type:     schema.TypeList,
										Optional: true,
										Elem:     &schema.Schema{Type: schema.TypeString},
									},
									"pod_affinity": {
										Type:     schema.TypeString,
										Optional: true,
									},
									"pod_anti_affinity": {
										Type:     schema.TypeString,
										Optional: true,
									},
									"node_affinity": {
										Type:     schema.TypeString,
										Optional: true,
									},
								},
							},
							MaxItems: 1,
						},
					},
				},
				MinItems: 1,
			},
			"bootstrap_scripts": {
				Type:     schema.TypeList,
				Optional: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"script_name": {
							Type:     schema.TypeString,
							Required: true,
						},
						"script_path": {
							Type:     schema.TypeString,
							Required: true,
						},
						"script_args": {
							Type:     schema.TypeString,
							Required: true,
						},
						"priority": {
							Type:       schema.TypeInt,
							Optional:   true,
							Deprecated: "Field 'priority' has been deprecated from provider version 1.227.0.",
						},
						"execution_moment": {
							Type:         schema.TypeString,
							Required:     true,
							ValidateFunc: StringInSlice([]string{"BEFORE_INSTALL", "AFTER_STARTED", "BEFORE_START"}, false),
						},
						"execution_fail_strategy": {
							Type:         schema.TypeString,
							Required:     true,
							ValidateFunc: validation.StringInSlice([]string{"FAILED_CONTINUE", "FAILED_BLOCK"}, false),
						},
						"node_selector": {
							Type:     schema.TypeSet,
							Required: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"node_select_type": {
										Type:         schema.TypeString,
										Required:     true,
										ValidateFunc: validation.StringInSlice([]string{"CLUSTER", "NODE_GROUP", "NODE"}, false),
									},
									"node_names": {
										Type:     schema.TypeList,
										Optional: true,
										Elem:     &schema.Schema{Type: schema.TypeString},
									},
									"node_group_id": {
										Type:       schema.TypeString,
										Optional:   true,
										Deprecated: "Field 'node_group_id' has been deprecated from provider version 1.227.0. New field 'node_group_ids' replaces it.",
									},
									"node_group_ids": {
										Type:     schema.TypeList,
										Optional: true,
										Elem:     &schema.Schema{Type: schema.TypeString},
									},
									"node_group_types": {
										Type:     schema.TypeList,
										Optional: true,
										Elem:     &schema.Schema{Type: schema.TypeString},
									},
									"node_group_name": {
										Type:       schema.TypeString,
										Optional:   true,
										Deprecated: "Field 'node_group_name' has been deprecated from provider version 1.227.0. New field 'node_group_names' replaces it.",
									},
									"node_group_names": {
										Type:     schema.TypeList,
										Optional: true,
										Elem:     &schema.Schema{Type: schema.TypeString},
									},
								},
							},
							MaxItems: 1,
						},
					},
				},
			},
			"tags": tagsSchema(),
		},
	}
}

func resourceAlicloudEmrV2ClusterCreate(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*connectivity.AliyunClient)
	var response map[string]interface{}
	action := "RunCluster"
	var err error
	createClusterRequest := map[string]interface{}{
		"RegionId": client.RegionId,
	}
	if v, ok := d.GetOk("resource_group_id"); ok {
		createClusterRequest["ResourceGroupId"] = v
	}

	if v, ok := d.GetOk("payment_type"); ok {
		createClusterRequest["PaymentType"] = v
	}

	if v, ok := d.GetOk("subscription_config"); ok {
		subscriptionConfig := v.(*schema.Set).List()
		if len(subscriptionConfig) == 1 {
			subscriptionConfigSource := subscriptionConfig[0].(map[string]interface{})
			subscriptionConfigMap := map[string]interface{}{
				"PaymentDurationUnit":   subscriptionConfigSource["payment_duration_unit"],
				"PaymentDuration":       subscriptionConfigSource["payment_duration"],
				"AutoRenew":             subscriptionConfigSource["auto_renew"],
				"AutoRenewDurationUnit": subscriptionConfigSource["auto_renew_duration_unit"],
				"AutoRenewDuration":     subscriptionConfigSource["auto_renew_duration"],
			}
			if value, exists := subscriptionConfigSource["auto_pay_order"]; exists {
				subscriptionConfigMap["AutoPayOrder"] = value.(bool)
			} else {
				subscriptionConfigMap["AutoPayOrder"] = true
			}

			createClusterRequest["SubscriptionConfig"] = convertMapToJsonStringIgnoreError(subscriptionConfigMap)
		}
	}

	if v, ok := d.GetOk("cluster_type"); ok {
		createClusterRequest["ClusterType"] = v
	}

	if v, ok := d.GetOk("release_version"); ok {
		createClusterRequest["ReleaseVersion"] = v
	}

	if v, ok := d.GetOk("cluster_name"); ok {
		createClusterRequest["ClusterName"] = v
	}

	if v, ok := d.GetOk("deploy_mode"); ok {
		createClusterRequest["DeployMode"] = v
	}

	if v, ok := d.GetOk("security_mode"); ok {
		createClusterRequest["SecurityMode"] = v
	}

	if v, ok := d.GetOk("log_collect_strategy"); ok && v.(string) != "" {
		createClusterRequest["LogCollectStrategy"] = v.(string)
	}

	if v, ok := d.GetOk("deletion_protection"); ok {
		createClusterRequest["DeletionProtection"] = v
	}

	applications := make([]map[string]interface{}, 0)
	if apps, ok := d.GetOk("applications"); ok {
		for _, application := range apps.(*schema.Set).List() {
			applications = append(applications, map[string]interface{}{"ApplicationName": application.(string)})
		}
	}
	createClusterRequest["Applications"], _ = convertListMapToJsonString(applications)

	applicationConfigs := make([]map[string]interface{}, 0)
	if appConfigs, ok := d.GetOk("application_configs"); ok {
		for _, appConfig := range appConfigs.(*schema.Set).List() {
			applicationConfig := map[string]interface{}{}
			kv := appConfig.(map[string]interface{})
			if v, ok := kv["application_name"]; ok {
				applicationConfig["ApplicationName"] = v
			}
			if v, ok := kv["config_file_name"]; ok {
				applicationConfig["ConfigFileName"] = v
			}
			if v, ok := kv["config_item_key"]; ok {
				applicationConfig["ConfigItemKey"] = v
			}
			if v, ok := kv["config_item_value"]; ok {
				applicationConfig["ConfigItemValue"] = v
			}
			if v, ok := kv["config_scope"]; ok {
				applicationConfig["ConfigScope"] = v
			}
			if v, ok := kv["node_group_name"]; ok {
				applicationConfig["NodeGroupName"] = v
			}
			if v, ok := kv["node_group_id"]; ok {
				applicationConfig["NodeGroupId"] = v
			}
			applicationConfigs = append(applicationConfigs, applicationConfig)
		}
	}
	createClusterRequest["ApplicationConfigs"], _ = convertListMapToJsonString(applicationConfigs)

	if v, ok := d.GetOk("node_attributes"); ok {
		nodeAttributes := v.(*schema.Set).List()
		if len(nodeAttributes) == 1 {
			nodeAttributesSource := nodeAttributes[0].(map[string]interface{})
			nodeAttributesSourceMap := map[string]interface{}{
				"VpcId":             nodeAttributesSource["vpc_id"],
				"ZoneId":            nodeAttributesSource["zone_id"],
				"SecurityGroupId":   nodeAttributesSource["security_group_id"],
				"RamRole":           nodeAttributesSource["ram_role"],
				"KeyPairName":       nodeAttributesSource["key_pair_name"],
				"DataDiskEncrypted": nodeAttributesSource["data_disk_encrypted"],
				"DataDiskKMSKeyId":  nodeAttributesSource["data_disk_kms_key_id"],
			}
			if value, exists := nodeAttributesSource["system_disk_encrypted"]; exists {
				nodeAttributesSourceMap["SystemDiskEncrypted"] = value.(bool)
			}
			if value, exists := nodeAttributesSource["system_disk_kms_key_id"]; exists && value.(string) != "" {
				nodeAttributesSourceMap["SystemDiskKMSKeyId"] = value.(string)
			}
			createClusterRequest["NodeAttributes"] = convertMapToJsonStringIgnoreError(nodeAttributesSourceMap)
		}
	}

	nodeGroups := make([]map[string]interface{}, 0)
	if nodeGroupsList, ok := d.GetOk("node_groups"); ok {
		for _, nodeGroupItem := range nodeGroupsList.([]interface{}) {
			nodeGroup := map[string]interface{}{}
			kv := nodeGroupItem.(map[string]interface{})
			if v, ok := kv["node_group_type"]; ok {
				nodeGroup["NodeGroupType"] = v
			}
			if v, ok := kv["node_group_name"]; ok {
				nodeGroup["NodeGroupName"] = v
			}
			if v, ok := kv["payment_type"]; ok {
				nodeGroup["PaymentType"] = v
			}
			if v, ok := kv["deployment_set_strategy"]; ok && v.(string) != "" {
				nodeGroup["DeploymentSetStrategy"] = v.(string)
			}
			if v, ok := kv["node_resize_strategy"]; ok && v.(string) != "" {
				nodeGroup["NodeResizeStrategy"] = v.(string)
			}
			if v, ok := kv["spot_strategy"]; ok && v.(string) != "" {
				nodeGroup["SpotStrategy"] = v.(string)
			}
			if v, ok := kv["subscription_config"]; ok {
				subscriptionConfigs := v.(*schema.Set).List()
				if len(subscriptionConfigs) == 1 {
					subscriptionConfig := map[string]interface{}{}
					subscriptionConfigMap := subscriptionConfigs[0].(map[string]interface{})
					if value, exists := subscriptionConfigMap["payment_duration_unit"]; exists {
						subscriptionConfig["PaymentDurationUnit"] = value
					}
					if value, exists := subscriptionConfigMap["payment_duration"]; exists {
						subscriptionConfig["PaymentDuration"] = value
					}
					if value, exists := subscriptionConfigMap["auto_renew"]; exists {
						subscriptionConfig["AutoRenew"] = value
					}
					if value, exists := subscriptionConfigMap["auto_renew_duration_unit"]; exists {
						subscriptionConfig["AutoRenewDurationUnit"] = value
					}
					if value, exists := subscriptionConfigMap["auto_renew_duration"]; exists {
						subscriptionConfig["AutoRenewDuration"] = value
					}
					if value, exists := subscriptionConfigMap["auto_pay_order"]; exists {
						subscriptionConfig["AutoPayOrder"] = value.(bool)
					} else {
						subscriptionConfig["AutoPayOrder"] = true
					}
					nodeGroup["SubscriptionConfig"] = subscriptionConfig
				}
			}
			if v, ok := kv["private_pool_options"]; ok {
				privatePoolOptions := v.(*schema.Set).List()
				if len(privatePoolOptions) == 1 {
					if value, exists := nodeGroup["PaymentType"]; exists && "PayAsYouGo" != value {
						return WrapError(Error("EMR cluster only 'PayAsYouGo' paymentType nodeGroup use private pool options."))
					}
					if value, exists := nodeGroup["node_group_type"]; exists && "TASK" != value {
						return WrapError(Error("EMR cluster only the node group type of 'TASK' use private pool options."))
					}

					privatePoolOption := map[string]interface{}{}
					privatePoolOptionMap := privatePoolOptions[0].(map[string]interface{})
					if value, exists := privatePoolOptionMap["private_pool_ids"]; exists {
						var ppIds []string
						for _, ppId := range value.(*schema.Set).List() {
							ppIds = append(ppIds, ppId.(string))
						}
						privatePoolOption["PrivatePoolIds"] = ppIds
					}
					if value, exists := privatePoolOptionMap["match_criteria"]; exists {
						privatePoolOption["MatchCriteria"] = value
					}
					nodeGroup["PrivatePoolOptions"] = privatePoolOption
				}
			}
			if v, ok := kv["spot_bid_prices"]; ok {
				spotBidPriceList := v.(*schema.Set).List()
				if len(spotBidPriceList) > 0 {
					spotBidPrices := make([]map[string]interface{}, 0)
					for _, spotBidPriceSource := range spotBidPriceList {
						spotBidPrice := map[string]interface{}{}
						spotBidPriceMap := spotBidPriceSource.(map[string]interface{})
						if value, exists := spotBidPriceMap["instance_type"]; exists {
							spotBidPrice["InstanceType"] = value
						}
						if value, exists := spotBidPriceMap["bid_price"]; exists {
							spotBidPrice["BidPrice"] = value
						}
						spotBidPrices = append(spotBidPrices, spotBidPrice)
					}
					nodeGroup["SpotBidPrices"] = spotBidPrices
				}
			}
			if v, ok := kv["vswitch_ids"]; ok {
				var vSwitchIds []string
				for _, vSwitchId := range v.(*schema.Set).List() {
					vSwitchIds = append(vSwitchIds, vSwitchId.(string))
				}
				nodeGroup["VSwitchIds"] = vSwitchIds
			}
			if v, ok := kv["with_public_ip"]; ok {
				nodeGroup["WithPublicIp"] = v
			}
			if v, ok := kv["additional_security_group_ids"]; ok {
				var additionalSecurityGroupIds []string
				for _, additionalSecurityGroupId := range v.(*schema.Set).List() {
					additionalSecurityGroupIds = append(additionalSecurityGroupIds, additionalSecurityGroupId.(string))
				}
				nodeGroup["AdditionalSecurityGroupIds"] = additionalSecurityGroupIds
			}
			if v, ok := kv["instance_types"]; ok {
				var instanceTypes []string
				for _, instanceType := range v.(*schema.Set).List() {
					instanceTypes = append(instanceTypes, instanceType.(string))
				}
				nodeGroup["InstanceTypes"] = instanceTypes
			}
			if v, ok := kv["node_count"]; ok {
				nodeGroup["NodeCount"] = v
			}
			if v, ok := kv["system_disk"]; ok {
				systemDisks := v.(*schema.Set).List()
				if len(systemDisks) == 1 {
					systemDisk := map[string]interface{}{}
					systemDiskMap := systemDisks[0].(map[string]interface{})
					if value, exists := systemDiskMap["category"]; exists {
						systemDisk["Category"] = value
					}
					if value, exists := systemDiskMap["size"]; exists {
						systemDisk["Size"] = value
					}
					if value, exists := systemDiskMap["performance_level"]; exists && value.(string) != "" {
						systemDisk["PerformanceLevel"] = value
					}
					if value, exists := systemDiskMap["count"]; exists {
						systemDisk["Count"] = value
					}
					nodeGroup["SystemDisk"] = systemDisk
				}
			}
			if v, ok := kv["data_disks"]; ok {
				dataDiskList := v.(*schema.Set).List()
				if len(dataDiskList) > 0 {
					dataDisks := make([]map[string]interface{}, 0)
					for _, dataDiskSource := range dataDiskList {
						dataDisk := map[string]interface{}{}
						dataDiskMap := dataDiskSource.(map[string]interface{})
						if value, exists := dataDiskMap["category"]; exists {
							dataDisk["Category"] = value
						}
						if value, exists := dataDiskMap["size"]; exists {
							dataDisk["Size"] = value
						}
						if value, exists := dataDiskMap["performance_level"]; exists && value.(string) != "" {
							dataDisk["PerformanceLevel"] = value
						}
						if value, exists := dataDiskMap["count"]; exists {
							dataDisk["Count"] = value
						}
						dataDisks = append(dataDisks, dataDisk)
					}
					nodeGroup["DataDisks"] = dataDisks
				}
			}
			if v, ok := kv["graceful_shutdown"]; ok {
				nodeGroup["GracefulShutdown"] = v
			}
			if v, ok := kv["spot_instance_remedy"]; ok {
				nodeGroup["SpotInstanceRemedy"] = v
			}
			if v, ok := kv["cost_optimized_config"]; ok {
				costOptimizedConfigs := v.(*schema.Set).List()
				if len(costOptimizedConfigs) == 1 {
					costOptimizedConfig := map[string]interface{}{}
					costOptimizedConfigMap := costOptimizedConfigs[0].(map[string]interface{})
					if value, exists := costOptimizedConfigMap["on_demand_base_capacity"]; exists {
						costOptimizedConfig["OnDemandBaseCapacity"] = value
					}
					if value, exists := costOptimizedConfigMap["on_demand_percentage_above_base_capacity"]; exists {
						costOptimizedConfig["OnDemandPercentageAboveBaseCapacity"] = value
					}
					if value, exists := costOptimizedConfigMap["spot_instance_pools"]; exists {
						costOptimizedConfig["SpotInstancePools"] = value
					}
					nodeGroup["CostOptimizedConfig"] = costOptimizedConfig
				}
			}
			if v, ok := kv["auto_scaling_policy"]; ok {
				scalingPolicies := v.([]interface{})
				if len(scalingPolicies) == 1 {
					nodeGroup["AutoScalingPolicy"] = adaptAutoScalingPolicyRequest(scalingPolicies[0].(map[string]interface{}))
				}
			}
			if v, ok := kv["ack_config"]; ok {
				ackConfig := v.([]interface{})
				if len(ackConfig) == 1 {
					nodeGroup["AckConfig"] = adaptAckConfigRequest(ackConfig[0].(map[string]interface{}))
					nodeGroup["IaaSType"] = "K8S"
					delete(nodeGroup, "InstanceTypes")
				}
			}
			nodeGroups = append(nodeGroups, nodeGroup)
		}
	}
	createClusterRequest["NodeGroups"], _ = convertListMapToJsonString(nodeGroups)

	if scripts, ok := d.GetOk("bootstrap_scripts"); ok {
		bootstrapScripts := make([]map[string]interface{}, 0)
		for _, script := range scripts.([]interface{}) {
			kv := script.(map[string]interface{})
			bootstrapScript := map[string]interface{}{}
			if v, ok := kv["script_name"]; ok {
				bootstrapScript["ScriptName"] = v
			}
			if v, ok := kv["script_path"]; ok {
				bootstrapScript["ScriptPath"] = v
			}
			if v, ok := kv["script_args"]; ok {
				bootstrapScript["ScriptArgs"] = v
			}
			if v, ok := kv["priority"]; ok {
				bootstrapScript["Priority"] = v
			}
			if v, ok := kv["execution_moment"]; ok {
				bootstrapScript["ExecutionMoment"] = v
			}
			if v, ok := kv["execution_fail_strategy"]; ok {
				bootstrapScript["ExecutionFailStrategy"] = v
			}
			if v, ok := kv["node_selector"]; ok {
				nodeSelectors := v.(*schema.Set).List()
				if len(nodeSelectors) == 1 {
					nodeSelector := map[string]interface{}{}
					nodeSelectorMap := nodeSelectors[0].(map[string]interface{})
					if value, exists := nodeSelectorMap["node_select_type"]; exists {
						nodeSelector["NodeSelectType"] = value
					}
					if value, exists := nodeSelectorMap["node_names"]; exists {
						var nodeNames []string
						for _, nodeName := range value.([]interface{}) {
							nodeNames = append(nodeNames, nodeName.(string))
						}
						nodeSelector["NodeNames"] = nodeNames
					}
					if value, exists := nodeSelectorMap["node_group_id"]; exists {
						nodeSelector["NodeGroupId"] = value
					}
					if value, exists := nodeSelectorMap["node_group_ids"]; exists {
						var nodeGroupIds []string
						for _, ngId := range value.([]interface{}) {
							nodeGroupIds = append(nodeGroupIds, ngId.(string))
						}
						nodeSelector["NodeGroupIds"] = nodeGroupIds
					}
					if value, exists := nodeSelectorMap["node_group_types"]; exists {
						var nodeGroupTypes []string
						for _, nodeGroupType := range value.([]interface{}) {
							nodeGroupTypes = append(nodeGroupTypes, nodeGroupType.(string))
						}
						nodeSelector["NodeGroupTypes"] = nodeGroupTypes
					}
					if value, exists := nodeSelectorMap["node_group_name"]; exists {
						nodeSelector["NodeGroupName"] = value
					}
					if value, exists := nodeSelectorMap["node_group_names"]; exists {
						var nodeGroupNames []string
						for _, ngName := range value.([]interface{}) {
							nodeGroupNames = append(nodeGroupNames, ngName.(string))
						}
						nodeSelector["NodeGroupNames"] = nodeGroupNames
					}
					bootstrapScript["NodeSelector"] = nodeSelector
				}
			}
			bootstrapScripts = append(bootstrapScripts, bootstrapScript)
		}
		createClusterRequest["BootstrapScripts"], _ = convertListMapToJsonString(bootstrapScripts)
	}

	if v, ok := d.GetOk("tags"); ok {
		tags := make([]map[string]interface{}, 0)
		for key, value := range v.(map[string]interface{}) {
			tags = append(tags, map[string]interface{}{
				"Key":   key,
				"Value": value,
			})
		}
		createClusterRequest["Tags"], _ = convertListMapToJsonString(tags)
	}

	wait := incrementalWait(3*time.Second, 5*time.Second)
	err = resource.Retry(d.Timeout(schema.TimeoutCreate), func() *resource.RetryError {
		response, err = client.RpcPost("Emr", "2021-03-20", action, nil, createClusterRequest, true)
		if err != nil {
			if NeedRetry(err) {
				wait()
				return resource.RetryableError(err)
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})
	addDebug(action, response, createClusterRequest)

	if err != nil {
		return WrapErrorf(err, DefaultErrorMsg, "alicloud_emrv2_cluster", action, AlibabaCloudSdkGoERROR)
	}

	d.SetId(fmt.Sprint(response["ClusterId"]))

	emrService := EmrService{client}
	stateConf := BuildStateConf([]string{"STARTING"}, []string{"RUNNING"}, d.Timeout(schema.TimeoutCreate),
		90*time.Second, emrService.EmrV2ClusterStateRefreshFunc(d.Id(), []string{"START_FAILED", "TERMINATED_WITH_ERRORS", "TERMINATED"}))
	stateConf.PollInterval = 10 * time.Second
	if _, err := stateConf.WaitForState(); err != nil {
		return WrapErrorf(err, IdMsg, d.Id())
	}

	return resourceAlicloudEmrV2ClusterRead(d, meta)
}

func resourceAlicloudEmrV2ClusterRead(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*connectivity.AliyunClient)
	emrService := EmrService{client}

	object, err := emrService.GetEmrV2Cluster(d.Id())

	if err != nil {
		if NotFoundError(err) {
			d.SetId("")
			return nil
		}
		return WrapError(err)
	}

	d.Set("cluster_name", object["ClusterName"])
	d.Set("cluster_type", object["ClusterType"])
	d.Set("payment_type", object["PaymentType"])
	d.Set("release_version", object["ReleaseVersion"])
	d.Set("deploy_mode", object["DeployMode"])
	d.Set("security_mode", object["SecurityMode"])
	d.Set("resource_group_id", object["ResourceGroupId"])
	d.Set("log_collect_strategy", object["LogCollectStrategy"])
	d.Set("deletion_protection", object["DeletionProtection"])

	if _, ok := object["SubscriptionConfig"]; ok {
		sc := d.Get("subscription_config").(*schema.Set).List()
		if len(sc) > 0 {
			d.Set("subscription_config", []map[string]interface{}{sc[0].(map[string]interface{})})
		}
	}

	var nodeAttributes []map[string]interface{}
	if v, ok := object["NodeAttributes"]; ok {
		nodeAttributesMap := v.(map[string]interface{})
		nodeAttribute := map[string]interface{}{
			"vpc_id":                nodeAttributesMap["VpcId"],
			"zone_id":               nodeAttributesMap["ZoneId"],
			"security_group_id":     nodeAttributesMap["SecurityGroupId"],
			"ram_role":              nodeAttributesMap["RamRole"],
			"key_pair_name":         nodeAttributesMap["KeyPairName"],
			"data_disk_encrypted":   nodeAttributesMap["DataDiskEncrypted"],
			"system_disk_encrypted": nodeAttributesMap["SystemDiskEncrypted"],
		}
		if v, exists := nodeAttributesMap["DataDiskKMSKeyId"]; exists && v.(string) != "" {
			nodeAttribute["data_disk_kms_key_id"] = v
		}
		if v, exists := nodeAttributesMap["SystemDiskKMSKeyId"]; exists && v.(string) != "" {
			nodeAttribute["system_disk_kms_key_id"] = v
		}

		oldNodeAttributes := d.Get("node_attributes")
		if oldNodeAttributes != nil && oldNodeAttributes.(*schema.Set).Len() > 0 {
			oldNodeAttributesMap := d.Get("node_attributes").(*schema.Set).List()[0].(map[string]interface{})
			if _, exists := oldNodeAttributesMap["system_disk_encrypted"]; exists {
				nodeAttribute["system_disk_encrypted"] = nodeAttributesMap["SystemDiskEncrypted"]
			}
			if value, exists := oldNodeAttributesMap["system_disk_kms_key_id"]; exists && value != "" {
				nodeAttribute["system_disk_kms_key_id"] = nodeAttributesMap["SystemDiskKMSKeyId"]
			}
		}

		nodeAttributes = append(nodeAttributes, nodeAttribute)
		d.Set("node_attributes", nodeAttributes)
	}
	var response map[string]interface{}
	action := "ListApplications"
	request := map[string]interface{}{
		"RegionId":  client.RegionId,
		"ClusterId": d.Id(),
	}
	wait := incrementalWait(3*time.Second, 3*time.Second)
	err = resource.Retry(5*time.Minute, func() *resource.RetryError {
		response, err = client.RpcPost("Emr", "2021-03-20", action, nil, request, true)
		if err != nil {
			if NeedRetry(err) {
				wait()
				return resource.RetryableError(err)
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})
	addDebug(action, response, request)
	if err != nil {
		return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
	}
	v, err := jsonpath.Get("$.Applications", response)
	if err != nil {
		return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.Applications", response)
	}
	if v != nil && len(v.([]interface{})) > 0 {
		emrInternalApp := map[string]struct{}{
			"TRE-EXTENSION":        {},
			"KERBEROS":             {},
			"METASTORE":            {},
			"MYSQL":                {},
			"SPARK-EXTENSION":      {},
			"SPARK-NATIVE":         {},
			"TAIHAODOCTOR":         {},
			"EMRHOOK":              {},
			"JINDOSDK":             {},
			"TRE":                  {},
			"EMRRUNTIME":           {},
			"EMRRUNTIME-EXTENSION": {},
		}
		var applications []string
		for _, item := range v.([]interface{}) {
			app := strings.ToUpper(item.(map[string]interface{})["ApplicationName"].(string))
			if _, ok := emrInternalApp[app]; ok {
				continue
			}
			applications = append(applications, app)
		}
		d.Set("applications", applications)
	}

	if v, ok := d.GetOk("application_configs"); ok && len(v.(*schema.Set).List()) > 0 {
		var applicationConfigs []map[string]interface{}
		for _, ac := range v.(*schema.Set).List() {
			applicationConfigs = append(applicationConfigs, ac.(map[string]interface{}))
		}
		d.Set("application_configs", applicationConfigs)
	}

	action = "ListScripts"
	listScriptsRequest := map[string]interface{}{
		"RegionId":   client.RegionId,
		"ClusterId":  d.Id(),
		"ScriptType": "BOOTSTRAP",
		"MaxResults": PageSizeLarge,
	}
	err = resource.Retry(5*time.Minute, func() *resource.RetryError {
		response, err = client.RpcPost("Emr", "2021-03-20", action, nil, listScriptsRequest, true)
		if err != nil {
			if NeedRetry(err) {
				wait()
				return resource.RetryableError(err)
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})
	addDebug(action, response, request)
	if err != nil {
		return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
	}
	v, err = jsonpath.Get("$.Scripts", response)
	if err != nil {
		return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.Scripts", response)
	}
	if v != nil && len(v.([]interface{})) > 0 {
		scriptsMaps := v.([]interface{})
		var scripts []map[string]interface{}
		for _, item := range scriptsMaps {
			scriptMap := item.(map[string]interface{})
			script := map[string]interface{}{
				"script_name":             scriptMap["ScriptName"],
				"script_path":             scriptMap["ScriptPath"],
				"script_args":             scriptMap["ScriptArgs"],
				"execution_moment":        scriptMap["ExecutionMoment"],
				"execution_fail_strategy": scriptMap["ExecutionFailStrategy"],
			}

			if v, ok := scriptMap["NodeSelector"]; ok && len(v.(map[string]interface{})) > 0 {
				nodeSelectorMap := v.(map[string]interface{})
				nodeSelector := map[string]interface{}{
					"node_select_type": nodeSelectorMap["NodeSelectType"],
				}

				if nodeGroupId, exists := nodeSelectorMap["NodeGroupId"]; exists && nodeGroupId.(string) != "" {
					nodeSelector["node_group_id"] = nodeGroupId
				}

				if ngIDs, exists := nodeSelectorMap["NodeGroupIds"]; exists && len(ngIDs.([]interface{})) > 0 {
					var nodeGroupIDs []string
					for _, ngID := range ngIDs.([]interface{}) {
						nodeGroupIDs = append(nodeGroupIDs, ngID.(string))
					}
					nodeSelector["node_group_ids"] = nodeGroupIDs
				}

				if nodeGroupName, exists := nodeSelectorMap["NodeGroupName"]; exists && nodeGroupName.(string) != "" {
					nodeSelector["node_group_name"] = nodeGroupName
				}

				if ngNames, exists := nodeSelectorMap["NodeGroupNames"]; exists && len(ngNames.([]interface{})) > 0 {
					var nodeGroupNames []string
					for _, ngName := range ngNames.([]interface{}) {
						nodeGroupNames = append(nodeGroupNames, ngName.(string))
					}
					nodeSelector["node_group_names"] = nodeGroupNames
				}

				if ngTypes, exists := nodeSelectorMap["NodeGroupTypes"]; exists && len(ngTypes.([]interface{})) > 0 {
					var nodeGroupTypes []string
					for _, ngType := range ngTypes.([]interface{}) {
						nodeGroupTypes = append(nodeGroupTypes, ngType.(string))
					}
					nodeSelector["node_group_types"] = nodeGroupTypes
				}

				if nn, exists := nodeSelectorMap["NodeNames"]; exists && len(nn.([]interface{})) > 0 {
					var nodeNames []string
					for _, nodeName := range nn.([]interface{}) {
						nodeNames = append(nodeNames, nodeName.(string))
					}
					nodeSelector["node_names"] = nodeNames
				}
				script["node_selector"] = []map[string]interface{}{nodeSelector}
			}
			scripts = append(scripts, script)
		}
		d.Set("bootstrap_scripts", scripts)
	}

	action = "ListNodeGroups"
	request["MaxResults"] = PageSizeLarge
	request["NodeGroupStates"] = []string{"RUNNING"}
	var nodeGroupObjects []interface{}

	for {
		err = resource.Retry(5*time.Minute, func() *resource.RetryError {
			response, err = client.RpcPost("Emr", "2021-03-20", action, nil, request, true)
			if err != nil {
				if NeedRetry(err) {
					wait()
					return resource.RetryableError(err)
				}
				return resource.NonRetryableError(err)
			}
			return nil
		})
		addDebug(action, response, request)
		if err != nil {
			return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
		}

		nodeGroupResp, err := jsonpath.Get("$.NodeGroups", response)
		if err != nil {
			return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.NodeGroups", response)
		}
		if nodeGroupResp != nil && len(nodeGroupResp.([]interface{})) > 0 {
			for _, ngr := range nodeGroupResp.([]interface{}) {
				nodeGroupObjects = append(nodeGroupObjects, ngr)
			}
			_, nextTokenExists := response["NextToken"]
			if len(nodeGroupResp.([]interface{})) < PageSizeLarge {
				break
			} else if len(nodeGroupResp.([]interface{})) == PageSizeLarge && !nextTokenExists {
				break
			}
		} else {
			break
		}

		nextToken, err := jsonpath.Get("$.NextToken", response)
		if err != nil {
			return WrapErrorf(err, FailedGetAttributeMsg, action, "$.NextToken", response)
		}
		request["NextToken"] = nextToken
	}
	if len(nodeGroupObjects) > 0 {
		v = nodeGroupObjects
	}
	delete(request, "NextToken")

	if v != nil && len(v.([]interface{})) > 0 {
		oldNodeGroupsMap := map[string]map[string]interface{}{}
		indexNodeGroupMap := map[string]int{}
		if oldNodeGroups, ok := d.GetOk("node_groups"); ok {
			for index, item := range oldNodeGroups.([]interface{}) {
				oldNodeGroup := item.(map[string]interface{})
				oldNodeGroupsMap[oldNodeGroup["node_group_name"].(string)] = oldNodeGroup
				indexNodeGroupMap[oldNodeGroup["node_group_name"].(string)] = index
			}
		}

		nodeGroupMaps := v.([]interface{})
		var nodeGroups []map[string]interface{}
		for _, item := range nodeGroupMaps {
			nodeGroupMap := item.(map[string]interface{})
			nodeGroup := map[string]interface{}{
				"node_group_type":      nodeGroupMap["NodeGroupType"],
				"node_group_name":      nodeGroupMap["NodeGroupName"],
				"payment_type":         nodeGroupMap["PaymentType"],
				"with_public_ip":       nodeGroupMap["WithPublicIp"],
				"graceful_shutdown":    nodeGroupMap["GracefulShutdown"],
				"spot_instance_remedy": nodeGroupMap["SpotInstanceRemedy"],
				"node_count":           nodeGroupMap["RunningNodeCount"],
			}

			if oldNodeGroup, exists := oldNodeGroupsMap[nodeGroupMap["NodeGroupName"].(string)]; exists {
				if v, ok := oldNodeGroup["subscription_config"]; ok {
					var subscriptionConfig []map[string]interface{}
					for _, item := range v.(*schema.Set).List() {
						subscriptionConfig = append(subscriptionConfig, item.(map[string]interface{}))
					}
					nodeGroup["subscription_config"] = subscriptionConfig
				}

				if v, ok := oldNodeGroup["spot_bid_prices"]; ok {
					var spotBidPrices []map[string]interface{}
					for _, item := range v.(*schema.Set).List() {
						spotBidPrices = append(spotBidPrices, item.(map[string]interface{}))
					}
					nodeGroup["spot_bid_prices"] = spotBidPrices
				}

				if v, ok := oldNodeGroup["auto_scaling_policy"]; ok && len(v.([]interface{})) == 1 {
					var scalingPolicies []map[string]interface{}
					scalingPolicy := map[string]interface{}{}
					scalingPolicyMap := v.([]interface{})[0].(map[string]interface{})
					if scalingPolicyValue, scalingPolicyExists := scalingPolicyMap["scaling_rules"]; scalingPolicyExists && len(scalingPolicyValue.([]interface{})) > 0 {
						var scalingRules []map[string]interface{}
						for _, sr := range scalingPolicyValue.([]interface{}) {
							scalingRule := map[string]interface{}{}
							scalingRuleMap := sr.(map[string]interface{})
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["rule_name"]; scalingRuleExists && scalingRuleValue.(string) != "" {
								scalingRule["rule_name"] = scalingRuleValue
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["trigger_type"]; scalingRuleExists && scalingRuleValue.(string) != "" {
								scalingRule["trigger_type"] = scalingRuleValue
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["activity_type"]; scalingRuleExists && scalingRuleValue.(string) != "" {
								scalingRule["activity_type"] = scalingRuleValue
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["adjustment_type"]; scalingRuleExists && scalingRuleValue.(string) != "" {
								scalingRule["adjustment_type"] = scalingRuleValue
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["adjustment_value"]; scalingRuleExists && scalingRuleValue.(int) != 0 {
								scalingRule["adjustment_value"] = scalingRuleValue
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["min_adjustment_value"]; scalingRuleExists {
								scalingRule["min_adjustment_value"] = scalingRuleValue
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["time_trigger"]; scalingRuleExists && len(scalingRuleValue.([]interface{})) > 0 {
								var timeTriggers []map[string]interface{}
								for _, tt := range scalingRuleValue.([]interface{}) {
									timeTrigger := map[string]interface{}{}
									timeTriggerMap := tt.(map[string]interface{})
									if timeTriggerValue, timeTriggerExists := timeTriggerMap["launch_time"]; timeTriggerExists && timeTriggerValue.(string) != "" {
										timeTrigger["launch_time"] = timeTriggerValue
									}
									if timeTriggerValue, timeTriggerExists := timeTriggerMap["start_time"]; timeTriggerExists && timeTriggerValue.(string) != "" {
										timeTrigger["start_time"] = timeTriggerValue
									}
									if timeTriggerValue, timeTriggerExists := timeTriggerMap["end_time"]; timeTriggerExists && timeTriggerValue.(string) != "" {
										timeTrigger["end_time"] = timeTriggerValue
									}
									if timeTriggerValue, timeTriggerExists := timeTriggerMap["launch_expiration_time"]; timeTriggerExists && timeTriggerValue.(int) != 0 {
										timeTrigger["launch_expiration_time"] = timeTriggerValue
									}
									if timeTriggerValue, timeTriggerExists := timeTriggerMap["recurrence_type"]; timeTriggerExists && timeTriggerValue.(string) != "" {
										timeTrigger["recurrence_type"] = timeTriggerValue
									}
									if timeTriggerValue, timeTriggerExists := timeTriggerMap["recurrence_value"]; timeTriggerExists && timeTriggerValue.(string) != "" {
										timeTrigger["recurrence_value"] = timeTriggerValue
									}
									timeTriggers = append(timeTriggers, timeTrigger)
								}
								scalingRule["time_trigger"] = timeTriggers
							}
							if scalingRuleValue, scalingRuleExists := scalingRuleMap["metrics_trigger"]; scalingRuleExists && len(scalingRuleValue.([]interface{})) > 0 {
								var metricsTriggers []map[string]interface{}
								for _, mt := range scalingRuleValue.([]interface{}) {
									metricsTrigger := map[string]interface{}{}
									metricsTriggerMap := mt.(map[string]interface{})
									if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["time_window"]; metricsTriggerExists && metricsTriggerValue.(int) != 0 {
										metricsTrigger["time_window"] = metricsTriggerValue
									}
									if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["evaluation_count"]; metricsTriggerExists && metricsTriggerValue.(int) != 0 {
										metricsTrigger["evaluation_count"] = metricsTriggerValue
									}
									if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["cool_down_interval"]; metricsTriggerExists && metricsTriggerValue.(int) != 0 {
										metricsTrigger["cool_down_interval"] = metricsTriggerValue
									}
									if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["condition_logic_operator"]; metricsTriggerExists && metricsTriggerValue.(string) != "" {
										metricsTrigger["condition_logic_operator"] = metricsTriggerValue
									}
									if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["time_constraints"]; metricsTriggerExists && len(metricsTriggerValue.([]interface{})) > 0 {
										var timeConstraints []map[string]interface{}
										for _, tc := range metricsTriggerValue.([]interface{}) {
											timeConstraint := map[string]interface{}{}
											timeConstraintMap := tc.(map[string]interface{})
											if timeConstraintValue, timeConstraintExists := timeConstraintMap["start_time"]; timeConstraintExists && timeConstraintValue.(string) != "" {
												timeConstraint["start_time"] = timeConstraintValue
											}
											if timeConstraintValue, timeConstraintExists := timeConstraintMap["end_time"]; timeConstraintExists && timeConstraintValue.(string) != "" {
												timeConstraint["end_time"] = timeConstraintValue
											}
											timeConstraints = append(timeConstraints, timeConstraint)
										}
										metricsTrigger["time_constraints"] = timeConstraints
									}
									if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["conditions"]; metricsTriggerExists && len(metricsTriggerValue.([]interface{})) > 0 {
										var conditions []map[string]interface{}
										for _, cd := range metricsTriggerValue.([]interface{}) {
											condition := map[string]interface{}{}
											conditionMap := cd.(map[string]interface{})
											if conditionValue, conditionExists := conditionMap["metric_name"]; conditionExists && conditionValue.(string) != "" {
												condition["metric_name"] = conditionValue
											}
											if conditionValue, conditionExists := conditionMap["statistics"]; conditionExists && conditionValue.(string) != "" {
												condition["statistics"] = conditionValue
											}
											if conditionValue, conditionExists := conditionMap["comparison_operator"]; conditionExists && conditionValue.(string) != "" {
												condition["comparison_operator"] = conditionValue
											}
											if conditionValue, conditionExists := conditionMap["threshold"]; conditionExists && conditionValue.(float64) != 0.0 {
												condition["threshold"] = conditionValue
											}
											if conditionValue, conditionExists := conditionMap["tags"]; conditionExists && len(conditionValue.([]interface{})) > 0 {
												var tags []map[string]interface{}
												for _, tg := range conditionValue.([]interface{}) {
													tag := map[string]interface{}{}
													tagMap := tg.(map[string]interface{})
													if tagValue, tagExists := tagMap["key"]; tagExists && tagValue.(string) != "" {
														tag["key"] = tagValue
													}
													if tagValue, tagExists := tagMap["value"]; tagExists && tagValue.(string) != "" {
														tag["value"] = tagValue
													}
													tags = append(tags, tag)
												}
												condition["tags"] = tags
											}
											conditions = append(conditions, condition)
										}
										metricsTrigger["conditions"] = conditions
									}
									metricsTriggers = append(metricsTriggers, metricsTrigger)
								}
								scalingRule["metrics_trigger"] = metricsTriggers
							}
							scalingRules = append(scalingRules, scalingRule)
						}
						scalingPolicy["scaling_rules"] = scalingRules
					}
					if scalingPolicyValue, scalingPolicyExists := scalingPolicyMap["constraints"]; scalingPolicyExists && len(scalingPolicyValue.([]interface{})) > 0 {
						var constraints []map[string]interface{}
						for _, ct := range scalingPolicyValue.([]interface{}) {
							constraint := map[string]interface{}{}
							constraintMap := ct.(map[string]interface{})
							if constraintValue, constraintExists := constraintMap["max_capacity"]; constraintExists && constraintValue.(int) != 0 {
								constraint["max_capacity"] = constraintValue
							}
							if constraintValue, constraintExists := constraintMap["min_capacity"]; constraintExists && constraintValue.(int) != 0 {
								constraint["min_capacity"] = constraintValue
							}
							constraints = append(constraints, constraint)
						}
						scalingPolicy["constraints"] = constraints
					}
					scalingPolicies = append(scalingPolicies, scalingPolicy)
					nodeGroup["auto_scaling_policy"] = scalingPolicies
				}
			}

			if v, ok := nodeGroupMap["VSwitchIds"]; ok && len(v.([]interface{})) > 0 {
				var vSwitchIDs []string
				for _, item := range v.([]interface{}) {
					vSwitchIDs = append(vSwitchIDs, item.(string))
				}
				nodeGroup["vswitch_ids"] = vSwitchIDs
			}

			if v, ok := nodeGroupMap["AdditionalSecurityGroupIds"]; ok && len(v.([]interface{})) > 0 {
				var additionalSecurityGroupIDs []string
				for _, item := range v.([]interface{}) {
					additionalSecurityGroupIDs = append(additionalSecurityGroupIDs, item.(string))
				}
				nodeGroup["additional_security_group_ids"] = additionalSecurityGroupIDs
			}

			if v, ok := nodeGroupMap["InstanceTypes"]; ok && len(v.([]interface{})) > 0 {
				var instanceTypes []string
				for _, item := range v.([]interface{}) {
					instanceTypes = append(instanceTypes, item.(string))
				}
				nodeGroup["instance_types"] = instanceTypes
			}

			if v, ok := nodeGroupMap["PrivatePoolOptions"]; ok {
				var privatePoolOptions []map[string]interface{}
				privatePoolOptionsMap := v.(map[string]interface{})
				if ppIds, ppIdsExists := privatePoolOptionsMap["PrivatePoolIds"]; ppIdsExists && len(ppIds.([]interface{})) > 0 {
					privatePoolOption := map[string]interface{}{}
					var privatePoolIDs []string
					for _, ppId := range ppIds.([]interface{}) {
						privatePoolIDs = append(privatePoolIDs, ppId.(string))
					}
					privatePoolOption["private_pool_ids"] = privatePoolIDs
					if mc, mcExists := privatePoolOptionsMap["MatchCriteria"]; mcExists && mc.(string) != "" {
						privatePoolOption["match_criteria"] = mc
					}
					privatePoolOptions = append(privatePoolOptions, privatePoolOption)
					nodeGroup["private_pool_options"] = privatePoolOptions
				}
			}

			if v, ok := nodeGroupMap["AckConfig"]; ok && len(v.(map[string]interface{})) > 0 {
				var ackConfigs []map[string]interface{}
				ackConfig := map[string]interface{}{}
				m := v.(map[string]interface{})
				ackConfig["ack_instance_id"] = m["AckInstanceId"]
				if value, exists := m["NodeSelectors"]; exists && len(value.([]interface{})) > 0 {
					var nodeSelectors []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						nodeSelectors = append(nodeSelectors, map[string]interface{}{
							"key":   ackConfigValue.(map[string]interface{})["Key"],
							"value": ackConfigValue.(map[string]interface{})["Value"],
						})
					}
					ackConfig["node_selectors"] = nodeSelectors
				}
				if value, exists := m["Tolerations"]; exists && len(value.([]interface{})) > 0 {
					var tolerations []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						tolerations = append(tolerations, map[string]interface{}{
							"key":      ackConfigValue.(map[string]interface{})["Key"],
							"value":    ackConfigValue.(map[string]interface{})["Value"],
							"operator": ackConfigValue.(map[string]interface{})["Operator"],
							"effect":   ackConfigValue.(map[string]interface{})["Effect"],
						})
					}
					ackConfig["tolerations"] = tolerations
				}
				if value, exists := m["Namespace"]; exists && value.(string) != "" {
					ackConfig["namespace"] = value
				}
				if value, exists := m["RequestCpu"]; exists {
					ackConfig["request_cpu"] = value
				}
				if value, exists := m["RequestMemory"]; exists {
					ackConfig["request_memory"] = value
				}
				if value, exists := m["LimitCpu"]; exists {
					ackConfig["limit_cpu"] = value
				}
				if value, exists := m["LimitMemory"]; exists {
					ackConfig["limit_memory"] = value
				}
				if value, exists := m["CustomLabels"]; exists && len(value.([]interface{})) > 0 {
					var customLabels []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						customLabels = append(customLabels, map[string]interface{}{
							"key":   ackConfigValue.(map[string]interface{})["Key"],
							"value": ackConfigValue.(map[string]interface{})["Value"],
						})
					}
					ackConfig["custom_labels"] = customLabels
				}
				if value, exists := m["CustomAnnotations"]; exists && len(value.([]interface{})) > 0 {
					var customAnnotations []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						customAnnotations = append(customAnnotations, map[string]interface{}{
							"key":   ackConfigValue.(map[string]interface{})["Key"],
							"value": ackConfigValue.(map[string]interface{})["Value"],
						})
					}
					ackConfig["custom_annotations"] = customAnnotations
				}
				if value, exists := m["Pvcs"]; exists && len(value.([]interface{})) > 0 {
					var pvcs []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						pvcs = append(pvcs, map[string]interface{}{
							"name":                    ackConfigValue.(map[string]interface{})["Name"],
							"path":                    ackConfigValue.(map[string]interface{})["Path"],
							"data_disk_storage_class": ackConfigValue.(map[string]interface{})["DataDiskStorageClass"],
							"data_disk_size":          ackConfigValue.(map[string]interface{})["DataDiskSize"],
						})
					}
					ackConfig["pvcs"] = pvcs
				}
				if value, exists := m["Volumes"]; exists && len(value.([]interface{})) > 0 {
					var volumes []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						volumes = append(volumes, map[string]interface{}{
							"name": ackConfigValue.(map[string]interface{})["Name"],
							"path": ackConfigValue.(map[string]interface{})["Path"],
							"type": ackConfigValue.(map[string]interface{})["Type"],
						})
					}
					ackConfig["volumes"] = volumes
				}
				if value, exists := m["VolumeMounts"]; exists && len(value.([]interface{})) > 0 {
					var volumeMounts []map[string]interface{}
					for _, ackConfigValue := range value.([]interface{}) {
						volumeMounts = append(volumeMounts, map[string]interface{}{
							"name": ackConfigValue.(map[string]interface{})["Name"],
							"path": ackConfigValue.(map[string]interface{})["Path"],
						})
					}
					ackConfig["volume_mounts"] = volumeMounts
				}
				if value, exists := m["PreStartCommand"]; exists && len(value.([]interface{})) > 0 {
					var preStartCommands []string
					for _, ackConfigValue := range value.([]interface{}) {
						preStartCommands = append(preStartCommands, ackConfigValue.(string))
					}
					ackConfig["pre_start_command"] = preStartCommands
				}
				if value, exists := m["PodAffinity"]; exists && value.(string) != "" {
					ackConfig["pod_affinity"] = value
				}
				if value, exists := m["PodAntiAffinity"]; exists && value.(string) != "" {
					ackConfig["pod_anti_affinity"] = value
				}
				if value, exists := m["NodeAffinity"]; exists && value.(string) != "" {
					ackConfig["node_affinity"] = value
				}
				ackConfigs = append(ackConfigs, ackConfig)
				nodeGroup["ack_config"] = ackConfigs

				if ong, exists := oldNodeGroupsMap[nodeGroupMap["NodeGroupName"].(string)]; exists {
					if instValue, instExists := ong["instance_types"]; instExists {
						var instanceTypes []string
						for _, item := range instValue.(*schema.Set).List() {
							instanceTypes = append(instanceTypes, item.(string))
						}
						nodeGroup["instance_types"] = instanceTypes
					}
				}
			}

			if oldNodeGroup, exists := oldNodeGroupsMap[nodeGroupMap["NodeGroupName"].(string)]; exists {
				if v, ok := oldNodeGroup["deployment_set_strategy"]; ok && v.(string) != "" {
					deploymentSetStrategy := v.(string)
					if st, stExists := nodeGroupMap["DeploymentSetStrategy"]; stExists && st.(string) != "" {
						deploymentSetStrategy = st.(string)
					}
					nodeGroup["deployment_set_strategy"] = deploymentSetStrategy
				}
			}

			if oldNodeGroup, exists := oldNodeGroupsMap[nodeGroupMap["NodeGroupName"].(string)]; exists {
				if v, ok := oldNodeGroup["node_resize_strategy"]; ok && v.(string) != "" {
					nodeGroup["node_resize_strategy"] = v.(string)
				}

				if v, ok := oldNodeGroup["spot_strategy"]; ok && v.(string) != "" {
					nodeGroup["spot_strategy"] = v.(string)
				}

				if v, ok := oldNodeGroup["cost_optimized_config"]; ok && len(v.(*schema.Set).List()) > 0 {
					var costOptimizedConfig []map[string]interface{}
					for _, coc := range v.(*schema.Set).List() {
						costOptimizedConfig = append(costOptimizedConfig, coc.(map[string]interface{}))
					}
					nodeGroup["cost_optimized_config"] = costOptimizedConfig
				}
			}

			if v, ok := nodeGroupMap["SystemDisk"]; ok {
				systemDiskMap := v.(map[string]interface{})
				systemDisk := map[string]interface{}{
					"category": systemDiskMap["Category"],
					"size":     formatInt(systemDiskMap["Size"]),
				}
				if oldNodeGroup, exists := oldNodeGroupsMap[nodeGroupMap["NodeGroupName"].(string)]; exists {
					oldSystemDisk := oldNodeGroup["system_disk"].(*schema.Set).List()[0].(map[string]interface{})
					if v, exists := oldSystemDisk["performance_level"]; exists && v != nil && v.(string) != "" {
						systemDisk["performance_level"] = v
					}
					if v, exists := oldSystemDisk["count"]; exists && v != nil {
						if count := formatInt(v); count > 0 {
							systemDisk["count"] = count
						}
					}
				}
				nodeGroup["system_disk"] = []map[string]interface{}{systemDisk}
			}

			if v, ok := nodeGroupMap["DataDisks"]; ok && len(v.([]interface{})) > 0 {
				var dataDisks []map[string]interface{}
				for _, item := range v.([]interface{}) {
					dataDisksMap := item.(map[string]interface{})
					dataDisk := map[string]interface{}{
						"category": dataDisksMap["Category"],
						"size":     formatInt(dataDisksMap["Size"]),
						"count":    formatInt(dataDisksMap["Count"]),
					}
					if oldNodeGroup, exists := oldNodeGroupsMap[nodeGroupMap["NodeGroupName"].(string)]; exists {
						oldDataDisk := oldNodeGroup["data_disks"].(*schema.Set).List()[0].(map[string]interface{})
						if v, exists := oldDataDisk["performance_level"]; exists && v != nil && v.(string) != "" {
							dataDisk["performance_level"] = v
						}
						if v, exists := oldDataDisk["count"]; exists && v != nil && v.(int) != 0 {
							if count := formatInt(v); count > 0 {
								dataDisk["count"] = formatInt(v)
							}
						}
					}
					dataDisks = append(dataDisks, dataDisk)
				}
				nodeGroup["data_disks"] = dataDisks
			}

			nodeGroups = append(nodeGroups, nodeGroup)
		}
		sort.Slice(nodeGroups, func(i, j int) bool {
			return indexNodeGroupMap[nodeGroups[i]["node_group_name"].(string)] < indexNodeGroupMap[nodeGroups[j]["node_group_name"].(string)]
		})

		d.Set("node_groups", nodeGroups)
	}

	tags, err := emrService.ListTagResourcesNew(d.Id(), string(TagResourceCluster))

	if err != nil {
		return WrapError(err)
	}
	d.Set("tags", tagsToMap(tags))

	return nil
}

func resourceAlicloudEmrV2ClusterUpdate(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*connectivity.AliyunClient)
	var err error
	var response map[string]interface{}
	d.Partial(true)
	emrService := EmrService{client}
	if err := emrService.SetEmrClusterTagsNew(d); err != nil {
		return WrapError(err)
	}

	if d.HasChange("cluster_name") || d.HasChange("log_collect_strategy") || d.HasChange("deletion_protection") {
		action := "UpdateClusterAttribute"
		request := map[string]interface{}{
			"ClusterId": d.Id(),
			"RegionId":  client.RegionId,
		}
		if d.HasChange("cluster_name") && d.Get("cluster_name").(string) != "" {
			request["ClusterName"] = d.Get("cluster_name")
		}
		if d.HasChange("log_collect_strategy") && d.Get("log_collect_strategy").(string) != "" {
			request["LogCollectStrategy"] = d.Get("log_collect_strategy")
		}
		if d.HasChange("deletion_protection") {
			request["DeletionProtection"] = d.Get("deletion_protection")
		}
		wait := incrementalWait(3*time.Second, 3*time.Second)
		err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
			response, err = client.RpcPost("Emr", "2021-03-20", action, nil, request, false)
			if err != nil {
				if NeedRetry(err) {
					wait()
					return resource.RetryableError(err)
				}
				return resource.NonRetryableError(err)
			}
			addDebug(action, response, request)
			return nil
		})
		if err != nil {
			return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
		}
		d.SetPartial("cluster_name")
	}

	if d.HasChange("payment_type") {
		_, newPaymentType := d.GetChange("payment_type")
		if "PayAsYouGo" == newPaymentType.(string) {
			return WrapError(Error("EMR cluster can only change paymentType from PayAsYouGo to Subscription."))
		}
		if !d.HasChange("node_groups") {
			return WrapError(Error("Subscription paymentType of emr cluster can not contains PayAsYouGo node group with 'MASTER' or 'CORE'."))
		}
	}

	if d.HasChange("node_groups") {
		oldNodeGroupsList, newNodeGroupsList := d.GetChange("node_groups")

		listNodeGroupsRequest := map[string]interface{}{
			"ClusterId":  d.Id(),
			"RegionId":   client.RegionId,
			"MaxResults": PageSizeLarge,
		}
		action := "ListNodeGroups"
		wait := incrementalWait(3*time.Second, 5*time.Second)
		err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
			response, err = client.RpcPost("Emr", "2021-03-20", action, nil, listNodeGroupsRequest, false)
			if err != nil {
				if NeedRetry(err) {
					wait()
					return resource.RetryableError(err)
				}
				return resource.NonRetryableError(err)
			}
			return nil
		})
		addDebug(action, response, listNodeGroupsRequest)
		if err != nil {
			return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
		}
		resp, err := jsonpath.Get("$.NodeGroups", response)

		if err != nil {
			return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.NodeGroups", response)
		}

		oldNodeGroupMap := map[string]map[string]interface{}{}
		for _, nodeGroupItem := range resp.([]interface{}) {
			oldNodeGroup := nodeGroupItem.(map[string]interface{})
			oldNodeGroupMap[oldNodeGroup["NodeGroupName"].(string)] = oldNodeGroup
		}

		originNodeGroupMap := map[string]map[string]interface{}{}
		for _, nodeGroupItem := range oldNodeGroupsList.([]interface{}) {
			originNodeGroup := nodeGroupItem.(map[string]interface{})
			originNodeGroupMap[originNodeGroup["node_group_name"].(string)] = originNodeGroup
		}

		newNodeGroupMap := map[string]map[string]interface{}{}
		for _, nodeGroupItem := range newNodeGroupsList.([]interface{}) {
			newNodeGroup := nodeGroupItem.(map[string]interface{})
			newNodeGroupMap[newNodeGroup["node_group_name"].(string)] = newNodeGroup
		}

		isUpdateClusterPaymentType := false
		if d.HasChange("payment_type") {
			action = "UpdateClusterPaymentType"
			updateClusterPaymentTypeRequest := map[string]interface{}{
				"ClusterId": d.Id(),
				"RegionId":  client.RegionId,
			}
			if sc, ok := d.GetOk("subscription_config"); ok {
				if autoPay, exists := sc.(*schema.Set).List()[0].(map[string]interface{})["auto_pay_order"]; exists {
					updateClusterPaymentTypeRequest["AutoPayOrder"] = autoPay.(bool)
				} else {
					updateClusterPaymentTypeRequest["AutoPayOrder"] = true
				}
				if autoRenew, exists := sc.(*schema.Set).List()[0].(map[string]interface{})["auto_renew"]; exists {
					updateClusterPaymentTypeRequest["AutoRenew"] = autoRenew.(bool)
				} else {
					updateClusterPaymentTypeRequest["AutoRenew"] = false
				}
			}
			var convertNodeGroups []map[string]interface{}
			for originNodeGroupName := range originNodeGroupMap {
				convertNodeGroup := map[string]interface{}{
					"NodeGroupId": oldNodeGroupMap[originNodeGroupName]["NodeGroupId"],
				}
				if newNodeGroup, ok := newNodeGroupMap[originNodeGroupName]; ok {
					if newNodeGroupValue, newNodeGroupExists := newNodeGroup["payment_type"]; newNodeGroupExists && "Subscription" == newNodeGroupValue {
						convertNodeGroup["PaymentType"] = newNodeGroupValue
					} else if "PayAsYouGo" == newNodeGroupValue && ("MASTER" == newNodeGroup["node_group_type"] || "CORE" == newNodeGroup["node_group_type"]) {
						return WrapError(Error("Subscription paymentType of emr cluster can not contains PayAsYouGo node group with 'MASTER' or 'CORE'."))
					} else {
						continue
					}
					if subscriptionConfig, exists := newNodeGroup["subscription_config"]; exists && len(subscriptionConfig.(*schema.Set).List()) > 0 {
						subscriptionConfigMap := subscriptionConfig.(*schema.Set).List()[0].(map[string]interface{})
						if subscriptionConfigValue, subscriptionConfigExists := subscriptionConfigMap["payment_duration"]; subscriptionConfigExists {
							convertNodeGroup["PaymentDuration"] = subscriptionConfigValue
						}
						if subscriptionConfigValue, subscriptionConfigExists := subscriptionConfigMap["payment_duration_unit"]; subscriptionConfigExists {
							convertNodeGroup["PaymentDurationUnit"] = subscriptionConfigValue
						}
					} else {
						return WrapError(Error("The '%s' nodeGroup: '%s' is needed parameter 'subscription_config' for changing paymentType.",
							newNodeGroup["node_group_type"], newNodeGroup["node_group_name"]))
					}
					convertNodeGroups = append(convertNodeGroups, convertNodeGroup)
				}
			}
			updateClusterPaymentTypeRequest["NodeGroups"] = convertNodeGroups
			wait = incrementalWait(3*time.Second, 5*time.Second)
			err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
				response, err = client.RpcPost("Emr", "2021-03-20", action, nil, updateClusterPaymentTypeRequest, false)
				if err != nil {
					if NeedRetry(err) {
						wait()
						return resource.RetryableError(err)
					}
					return resource.NonRetryableError(err)
				}
				return nil
			})
			addDebug(action, response, updateClusterPaymentTypeRequest)
			if err != nil {
				return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
			}
			// Wait for cluster payment type has been changed
			if err = resource.Retry(5*time.Minute, func() *resource.RetryError {
				if cluster, err := emrService.GetEmrV2Cluster(d.Id()); err != nil {
					return resource.NonRetryableError(err)
				} else if cluster["PaymentType"].(string) == "Subscription" {
					return nil
				}
				return resource.RetryableError(Error("Waiting for cluster %s payment type to be changed.", d.Id()))
			}); err != nil {
				return WrapError(err)
			}
			isUpdateClusterPaymentType = true
		}

		var increaseNodesGroups []map[string]interface{}
		var decreaseNodesGroups []map[string]interface{}

		for nodeGroupName, newNodeGroup := range newNodeGroupMap {
			if oldNodeGroup, ok := oldNodeGroupMap[nodeGroupName]; ok {
				if !reflect.DeepEqual(originNodeGroupMap[nodeGroupName]["auto_scaling_policy"], newNodeGroup["auto_scaling_policy"]) {
					removeScalingPolicy := false
					if len(newNodeGroup["auto_scaling_policy"].([]interface{})) > 0 {
						adaptedScalingPolicy := adaptAutoScalingPolicyRequest(newNodeGroup["auto_scaling_policy"].([]interface{})[0].(map[string]interface{}))
						rulesExists := false
						constraintsExists := false
						if aspValue, aspExists := adaptedScalingPolicy["scalingRules"]; aspExists {
							rulesExists = aspExists
							adaptedScalingPolicy["ScalingRules"] = aspValue
							delete(adaptedScalingPolicy, "scalingRules")
						}
						if aspValue, aspExists := adaptedScalingPolicy["constraints"]; aspExists {
							constraintsExists = aspExists
							adaptedScalingPolicy["Constraints"] = aspValue
							delete(adaptedScalingPolicy, "constraints")
						}
						if rulesExists || constraintsExists {
							adaptedScalingPolicy["RegionId"] = client.RegionId
							adaptedScalingPolicy["ClusterId"] = d.Id()
							adaptedScalingPolicy["NodeGroupId"] = oldNodeGroup["NodeGroupId"]

							action = "PutAutoScalingPolicy"
							err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
								response, err = client.RpcPost("Emr", "2021-03-20", action, nil, adaptedScalingPolicy, false)
								if err != nil {
									if NeedRetry(err) {
										wait()
										return resource.RetryableError(err)
									}
									return resource.NonRetryableError(err)
								}
								return nil
							})
							addDebug(action, response, adaptedScalingPolicy)
							if err != nil {
								return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
							}
						} else {
							removeScalingPolicy = true
						}
					} else {
						removeScalingPolicy = true
					}
					if removeScalingPolicy {
						removeScalingPolicyRequest := map[string]interface{}{
							"RegionId":    client.RegionId,
							"ClusterId":   d.Id(),
							"NodeGroupId": oldNodeGroup["NodeGroupId"],
						}
						action = "RemoveAutoScalingPolicy"
						err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
							response, err = client.RpcPost("Emr", "2021-03-20", action, nil, removeScalingPolicyRequest, false)
							if err != nil {
								if NeedRetry(err) {
									wait()
									return resource.RetryableError(err)
								}
								return resource.NonRetryableError(err)
							}
							return nil
						})
						addDebug(action, response, removeScalingPolicyRequest)
						if err != nil {
							return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
						}
					}
				}
				if !reflect.DeepEqual(originNodeGroupMap[nodeGroupName]["ack_config"], newNodeGroup["ack_config"]) && "K8S" == oldNodeGroup["IaaSType"] {
					ackConfigs := newNodeGroup["ack_config"].([]interface{})
					if len(ackConfigs) > 0 {
						updateNodeGroupAttributesRequest := map[string]interface{}{
							"RegionId":    client.RegionId,
							"ClusterId":   d.Id(),
							"NodeGroupId": oldNodeGroup["NodeGroupId"],
							"AckConfig":   adaptAckConfigRequest(ackConfigs[0].(map[string]interface{})),
						}
						action = "UpdateNodeGroupAttributes"
						err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
							response, err = client.RpcPost("Emr", "2021-03-20", action, nil, updateNodeGroupAttributesRequest, false)
							if err != nil {
								if NeedRetry(err) {
									wait()
									return resource.RetryableError(err)
								}
								return resource.NonRetryableError(err)
							}
							return nil
						})
						addDebug(action, response, updateNodeGroupAttributesRequest)
						if err != nil {
							return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
						}
					}
				}

				newNodeCount := formatInt(newNodeGroup["node_count"])
				oldNodeCount := formatInt(oldNodeGroup["RunningNodeCount"])

				// increase nodes
				if oldNodeCount < newNodeCount {
					if "MASTER" == newNodeGroup["node_group_type"].(string) {
						return WrapError(Error("EMR cluster can not increase the node group type of ['MASTER']."))
					}
					count := newNodeCount - oldNodeCount
					increaseNodesGroup := map[string]interface{}{}
					increaseNodesGroup["RegionId"] = client.RegionId
					increaseNodesGroup["ClusterId"] = d.Id()
					increaseNodesGroup["NodeGroupId"] = oldNodeGroup["NodeGroupId"]
					increaseNodesGroup["IncreaseNodeCount"] = count
					increaseNodesGroup["AutoRenew"] = false
					if "Subscription" == newNodeGroup["payment_type"].(string) {
						subscriptionConfig := newNodeGroup["subscription_config"].(*schema.Set).List()
						if len(subscriptionConfig) == 1 {
							configMap := subscriptionConfig[0].(map[string]interface{})
							increaseNodesGroup["PaymentDuration"] = configMap["payment_duration"]
							increaseNodesGroup["PaymentDurationUnit"] = configMap["payment_duration_unit"]
							if value, exists := configMap["auto_pay_order"]; exists {
								increaseNodesGroup["AutoPayOrder"] = value.(bool)
							} else {
								increaseNodesGroup["AutoPayOrder"] = true
							}
							if value, exists := configMap["auto_renew"]; exists {
								increaseNodesGroup["AutoRenew"] = value.(bool)
							}
						}
					}
					increaseNodesGroups = append(increaseNodesGroups, increaseNodesGroup)
				} else if oldNodeCount > newNodeCount { // decrease nodes
					// EMR cluster can only decrease 'TASK, GATEWAY' node group.
					nodeGroupType := newNodeGroup["node_group_type"].(string)
					if "TASK" != nodeGroupType && "GATEWAY" != nodeGroupType {
						return WrapError(Error("EMR cluster can only decrease the node group type of ['TASK', 'GATEWAY']."))
					}
					decreaseNodesGroup := map[string]interface{}{
						"ClusterId":         d.Id(),
						"RegionId":          client.RegionId,
						"DecreaseNodeCount": oldNodeCount - newNodeCount,
						"NodeGroupId":       oldNodeGroup["NodeGroupId"],
					}
					decreaseNodesGroups = append(decreaseNodesGroups, decreaseNodesGroup)
				}

				// increase node disk size, we can only support single disk type.
				currDataDisk := oldNodeGroup["DataDisks"].([]interface{})[0].(map[string]interface{})
				targetDataDisk := newNodeGroup["data_disks"].(*schema.Set).List()[0].(map[string]interface{})
				if formatInt(targetDataDisk["size"]) < formatInt(currDataDisk["Size"]) {
					return WrapError(Error("EMR cluster can only increase node disk, decrease node disk is not supported."))
				} else if formatInt(targetDataDisk["size"]) > formatInt(currDataDisk["Size"]) {
					if currDataDisk["Category"].(string) == "local_hdd_pro" {
						return WrapError(Error("EMR cluster can not support increase node disk with 'local_hdd_pro' disk type."))
					}
					action := "IncreaseNodesDiskSize"
					increaseNodeDiskSizeRequest := map[string]interface{}{
						"ClusterId":   d.Id(),
						"RegionId":    client.RegionId,
						"NodeGroupId": oldNodeGroup["NodeGroupId"],
						"DataDiskSizes": []map[string]interface{}{
							{
								"Category": currDataDisk["Category"].(string),
								"Size":     formatInt(targetDataDisk["size"]),
							},
						},
					}
					wait := incrementalWait(3*time.Second, 5*time.Second)
					err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
						response, err = client.RpcPost("Emr", "2021-03-20", action, nil, increaseNodeDiskSizeRequest, false)
						if err != nil {
							if NeedRetry(err) {
								wait()
								return resource.RetryableError(err)
							}
							return resource.NonRetryableError(err)
						}
						return nil
					})
					addDebug(action, response, increaseNodeDiskSizeRequest)
					if err != nil {
						return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
					}
				}
			} else { // 'Task' NodeGroupType may not exist when create emr_cluster
				subscriptionConfig := map[string]interface{}{}
				if "Subscription" == newNodeGroup["payment_type"] {
					subscriptionMap := newNodeGroup["subscription_config"].(*schema.Set).List()[0].(map[string]interface{})
					subscriptionConfig["PaymentDurationUnit"] = subscriptionMap["payment_duration_unit"]
					subscriptionConfig["PaymentDuration"] = subscriptionMap["payment_duration"]
					subscriptionConfig["AutoRenewDurationUnit"] = subscriptionMap["auto_renew_duration_unit"]
					subscriptionConfig["AutoRenewDuration"] = subscriptionMap["auto_renew_duration"]
					subscriptionConfig["AutoRenew"] = false
					if value, exists := subscriptionMap["auto_pay_order"]; exists {
						subscriptionConfig["AutoPayOrder"] = value.(bool)
					} else {
						subscriptionConfig["AutoPayOrder"] = true
					}
					if value, exists := subscriptionMap["auto_renew"]; exists {
						subscriptionConfig["AutoRenew"] = value.(bool)
					}
				}
				var spotBidPrices []map[string]interface{}
				for _, v := range newNodeGroup["spot_bid_prices"].(*schema.Set).List() {
					sbpMap := v.(map[string]interface{})
					spotBidPrices = append(spotBidPrices, map[string]interface{}{
						"InstanceType": sbpMap["instance_type"],
						"BidPrice":     sbpMap["bid_price"],
					})
				}
				systemDiskMap := newNodeGroup["system_disk"].(*schema.Set).List()[0].(map[string]interface{})
				var dataDisks []map[string]interface{}
				for _, v := range newNodeGroup["data_disks"].(*schema.Set).List() {
					dataDiskMap := v.(map[string]interface{})
					dataDisk := map[string]interface{}{
						"Category": dataDiskMap["category"],
						"Size":     dataDiskMap["size"],
						"Count":    dataDiskMap["count"],
					}
					if value, exists := dataDiskMap["performance_level"]; exists && value.(string) != "" {
						dataDisk["PerformanceLevel"] = value.(string)
					}
					dataDisks = append(dataDisks, dataDisk)
				}
				nodeGroupParam := map[string]interface{}{
					"NodeGroupType":      newNodeGroup["node_group_type"],
					"NodeGroupName":      nodeGroupName,
					"PaymentType":        newNodeGroup["payment_type"],
					"SubscriptionConfig": subscriptionConfig,
					"SpotBidPrices":      spotBidPrices,
					"WithPublicIp":       newNodeGroup["with_public_ip"],
					"NodeCount":          newNodeGroup["node_count"],
					"DataDisks":          dataDisks,
					"GracefulShutdown":   newNodeGroup["graceful_shutdown"],
					"SpotInstanceRemedy": newNodeGroup["spot_instance_remedy"],
				}
				if value, exists := newNodeGroup["private_pool_options"]; exists && len(value.(*schema.Set).List()) > 0 {
					if value, exists := newNodeGroup["payment_type"]; exists && "PayAsYouGo" != value {
						return WrapError(Error("EMR cluster only 'PayAsYouGo' paymentType nodeGroup use private pool options."))
					}
					if value, exists := newNodeGroup["node_group_type"]; exists && "TASK" != value {
						return WrapError(Error("EMR cluster only the node group type of 'TASK' use private pool options."))
					}

					privatePoolOptions := map[string]interface{}{}
					privatePoolOptionsMap := value.(*schema.Set).List()[0].(map[string]interface{})
					if ppIds, ppIdsExists := privatePoolOptionsMap["private_pool_ids"]; ppIdsExists && len(ppIds.(*schema.Set).List()) > 0 {
						var privatePoolIds []string
						for _, ppId := range ppIds.(*schema.Set).List() {
							privatePoolIds = append(privatePoolIds, ppId.(string))
						}
						privatePoolOptions["PrivatePoolIds"] = privatePoolIds
					}
					if mm, mmExists := privatePoolOptionsMap["match_criteria"]; mmExists && mm.(string) != "" {
						privatePoolOptions["MatchCriteria"] = mm
					}
					nodeGroupParam["PrivatePoolOptions"] = privatePoolOptions
				}
				if value, exists := newNodeGroup["auto_scaling_policy"]; exists && len(value.([]interface{})) > 0 {
					nodeGroupParam["AutoScalingPolicy"] = adaptAutoScalingPolicyRequest(value.([]interface{})[0].(map[string]interface{}))
				}
				if value, exists := newNodeGroup["deployment_set_strategy"]; exists && value.(string) != "" {
					nodeGroupParam["DeploymentSetStrategy"] = value.(string)
				}
				if value, exists := newNodeGroup["node_resize_strategy"]; exists && value.(string) != "" {
					nodeGroupParam["NodeResizeStrategy"] = value.(string)
				}
				if value, exists := newNodeGroup["spot_strategy"]; exists && value.(string) != "" {
					nodeGroupParam["SpotStrategy"] = value.(string)
				}
				vSwitchIDList := newNodeGroup["vswitch_ids"].(*schema.Set).List()
				if len(vSwitchIDList) > 0 {
					var vSwitchIDs []string
					for _, vSwitchID := range vSwitchIDList {
						vSwitchIDs = append(vSwitchIDs, vSwitchID.(string))
					}
					nodeGroupParam["VSwitchIds"] = vSwitchIDs
				}
				systemDisk := map[string]interface{}{
					"Category": systemDiskMap["category"],
					"Size":     systemDiskMap["size"],
					"Count":    systemDiskMap["count"],
				}
				if value, exists := systemDiskMap["performance_level"]; exists && value.(string) != "" {
					systemDisk["PerformanceLevel"] = value.(string)
				}
				nodeGroupParam["SystemDisk"] = systemDisk

				instanceTypeList := newNodeGroup["instance_types"].(*schema.Set).List()
				if len(instanceTypeList) > 0 {
					var instanceTypes []string
					for _, instanceType := range instanceTypeList {
						instanceTypes = append(instanceTypes, instanceType.(string))
					}
					nodeGroupParam["InstanceTypes"] = instanceTypes
				}
				if value, exists := newNodeGroup["ack_config"]; exists && len(value.([]interface{})) > 0 {
					nodeGroupParam["AckConfig"] = adaptAckConfigRequest(value.([]interface{})[0].(map[string]interface{}))
					nodeGroupParam["IaaSType"] = "K8S"
					delete(nodeGroupParam, "InstanceTypes")
				}

				addSecurityGroupIDList := newNodeGroup["additional_security_group_ids"].(*schema.Set).List()
				if len(addSecurityGroupIDList) > 0 {
					var addSecurityGroupIDs []string
					for _, addSecurityGroupID := range addSecurityGroupIDList {
						addSecurityGroupIDs = append(addSecurityGroupIDs, addSecurityGroupID.(string))
					}
					nodeGroupParam["AdditionalSecurityGroupIds"] = addSecurityGroupIDs
				}

				costOptimizedConfigList := newNodeGroup["cost_optimized_config"].(*schema.Set).List()
				if len(costOptimizedConfigList) > 0 {
					costOptimizedConfig := costOptimizedConfigList[0].(map[string]interface{})
					nodeGroupParam["CostOptimizedConfig"] = map[string]interface{}{
						"OnDemandBaseCapacity":                costOptimizedConfig["on_demand_base_capacity"],
						"OnDemandPercentageAboveBaseCapacity": costOptimizedConfig["on_demand_percentage_above_base_capacity"],
						"SpotInstancePools":                   costOptimizedConfig["spot_instance_pools"],
					}
				}
				createNodeGroupRequest := map[string]interface{}{
					"ClusterId": d.Id(),
					"RegionId":  client.RegionId,
					"NodeGroup": nodeGroupParam,
				}

				action = "CreateNodeGroup"
				err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
					response, err = client.RpcPost("Emr", "2021-03-20", action, nil, createNodeGroupRequest, false)
					if err != nil {
						if NeedRetry(err) {
							wait()
							return resource.RetryableError(err)
						}
						return resource.NonRetryableError(err)
					}
					return nil
				})
				addDebug(action, response, createNodeGroupRequest)
				if err != nil {
					return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
				}

				listNodeGroupsRequest := map[string]interface{}{
					"ClusterId":      d.Id(),
					"NodeGroupNames": []string{nodeGroupName},
					"RegionId":       client.RegionId,
				}
				action = "ListNodeGroups"
				err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
					response, err = client.RpcPost("Emr", "2021-03-20", action, nil, listNodeGroupsRequest, false)
					if err != nil {
						if NeedRetry(err) {
							wait()
							return resource.RetryableError(err)
						}
						return resource.NonRetryableError(err)
					}
					return nil
				})
				addDebug(action, response, listNodeGroupsRequest)
				if err != nil {
					return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
				}
				resp, err := jsonpath.Get("$.NodeGroups", response)
				if err != nil {
					return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.NodeGroups", response)
				}

				if len(resp.([]interface{})) == 0 {
					continue
				}

				nodeGroupId := resp.([]interface{})[0].(map[string]interface{})["NodeGroupId"].(string)

				newNodeCount := formatInt(newNodeGroup["node_count"])
				if newNodeCount > 0 {
					increaseNodesGroup := map[string]interface{}{}
					increaseNodesGroup["RegionId"] = client.RegionId
					increaseNodesGroup["ClusterId"] = d.Id()
					increaseNodesGroup["NodeGroupId"] = nodeGroupId
					increaseNodesGroup["IncreaseNodeCount"] = newNodeCount
					increaseNodesGroup["AutoPayOrder"] = true
					increaseNodesGroup["AutoRenew"] = false
					if "Subscription" == newNodeGroup["payment_type"].(string) {
						subscriptionConfig := newNodeGroup["subscription_config"].(*schema.Set).List()
						if len(subscriptionConfig) == 1 {
							configMap := subscriptionConfig[0].(map[string]interface{})
							increaseNodesGroup["PaymentDuration"] = configMap["payment_duration"]
							increaseNodesGroup["PaymentDurationUnit"] = configMap["payment_duration_unit"]
							if value, exists := configMap["auto_pay_order"]; exists {
								increaseNodesGroup["AutoPayOrder"] = value.(bool)
							}
							if value, exists := configMap["auto_renew"]; exists {
								increaseNodesGroup["AutoRenew"] = value.(bool)
							}
						}
					}
					increaseNodesGroups = append(increaseNodesGroups, increaseNodesGroup)
				}
			}
		}

		var deleteNodeGroups []map[string]interface{}
		for nodeGroupName, oldNodeGroup := range oldNodeGroupMap { // Delete empty nodeGroup
			if newNodeGroup, ok := newNodeGroupMap[nodeGroupName]; !ok {
				oldNodeCount := formatInt(oldNodeGroup["RunningNodeCount"])
				if oldNodeCount > 0 {
					return WrapError(Error("The [nodeGroup: %v, nodeGroupType: %v] can not delete cause exists running nodes", nodeGroupName, oldNodeGroup["NodeGroupType"].(string)))
				}
				deleteNodeGroups = append(deleteNodeGroups, map[string]interface{}{
					"ClusterId":   d.Id(),
					"NodeGroupId": oldNodeGroup["NodeGroupId"],
					"RegionId":    client.RegionId,
				})
			} else if newNodeGroup["payment_type"] == "Subscription" && oldNodeGroup["PaymentType"] == "PayAsYouGo" &&
				!isUpdateClusterPaymentType && d.Get("payment_type") == "Subscription" {
				action = "UpdateNodeGroupPaymentType"
				UpdateNodeGroupPaymentTypeRequest := map[string]interface{}{
					"ClusterId": d.Id(),
					"RegionId":  client.RegionId,
				}
				updateNodeGroupPaymentType := map[string]interface{}{
					"NodeGroupId": oldNodeGroup["NodeGroupId"],
				}
				if newNodeGroupValue, newNodeGroupExists := newNodeGroup["payment_type"]; newNodeGroupExists {
					updateNodeGroupPaymentType["PaymentType"] = newNodeGroupValue
				}
				if subscriptionConfig, exists := newNodeGroup["subscription_config"]; exists && len(subscriptionConfig.(*schema.Set).List()) > 0 {
					subscriptionConfigMap := subscriptionConfig.(*schema.Set).List()[0].(map[string]interface{})
					if subscriptionConfigValue, subscriptionConfigExists := subscriptionConfigMap["payment_duration"]; subscriptionConfigExists {
						updateNodeGroupPaymentType["PaymentDuration"] = subscriptionConfigValue
					}
					if subscriptionConfigValue, subscriptionConfigExists := subscriptionConfigMap["payment_duration_unit"]; subscriptionConfigExists {
						updateNodeGroupPaymentType["PaymentDurationUnit"] = subscriptionConfigValue
					}
					if subscriptionConfigValue, subscriptionConfigExists := subscriptionConfigMap["auto_pay_order"]; subscriptionConfigExists {
						UpdateNodeGroupPaymentTypeRequest["AutoPayOrder"] = subscriptionConfigValue
					} else {
						UpdateNodeGroupPaymentTypeRequest["AutoPayOrder"] = true
					}
					if subscriptionConfigValue, subscriptionConfigExists := subscriptionConfigMap["auto_renew"]; subscriptionConfigExists {
						UpdateNodeGroupPaymentTypeRequest["AutoRenew"] = subscriptionConfigValue
					} else {
						UpdateNodeGroupPaymentTypeRequest["AutoRenew"] = false
					}
				} else {
					return WrapError(Error("The '%s' nodeGroup: '%s' is needed parameter 'subscription_config' for changing paymentType.",
						newNodeGroup["node_group_type"], newNodeGroup["node_group_name"]))
				}
				UpdateNodeGroupPaymentTypeRequest["NodeGroup"] = updateNodeGroupPaymentType
				wait = incrementalWait(3*time.Second, 5*time.Second)
				err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
					response, err = client.RpcPost("Emr", "2021-03-20", action, nil, UpdateNodeGroupPaymentTypeRequest, false)
					if err != nil {
						if NeedRetry(err) {
							wait()
							return resource.RetryableError(err)
						}
						return resource.NonRetryableError(err)
					}
					return nil
				})
				addDebug(action, response, UpdateNodeGroupPaymentTypeRequest)
				if err != nil {
					return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
				}
				// Wait for node group payment type has been changed
				if err = resource.Retry(5*time.Minute, func() *resource.RetryError {
					nodeGroupId := updateNodeGroupPaymentType["NodeGroupId"].(string)
					if nodeGroups, err := emrService.ListEmrV2NodeGroups(d.Id(), []string{nodeGroupId}); err != nil {
						return resource.NonRetryableError(err)
					} else if len(nodeGroups) > 0 && "Subscription" == nodeGroups[0].(map[string]interface{})["PaymentType"].(string) {
						return nil
					}
					return resource.RetryableError(Error("Waiting for node group %s payment type to be changed.", nodeGroupId))
				}); err != nil {
					return WrapError(err)
				}
			} else if !isUpdateClusterPaymentType && d.Get("payment_type") == "Subscription" &&
				oldNodeGroup["PaymentType"] == "Subscription" && newNodeGroup["payment_type"] == "PayAsYouGo" {
				return WrapError(Error("EMR cluster can only change paymentType from PayAsYouGo to Subscription."))
			}
		}

		var wg sync.WaitGroup
		var cm sync.Map
		waitFlag := false
		for _, increaseNodesGroupRequest := range increaseNodesGroups {
			waitFlag = true
			action := "IncreaseNodes"
			wait := incrementalWait(3*time.Second, 5*time.Second)
			err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
				response, err = client.RpcPost("Emr", "2021-03-20", action, nil, increaseNodesGroupRequest, false)
				if err != nil {
					if NeedRetry(err) {
						wait()
						return resource.RetryableError(err)
					}
					return resource.NonRetryableError(err)
				}
				return nil
			})
			addDebug(action, response, increaseNodesGroupRequest)
			if err != nil {
				return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
			}
			resp, err := jsonpath.Get("$.OperationId", response)
			if err != nil {
				return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.OperationId", response)
			}
			wg.Add(1)
			go emrService.WaitForEmrV2Operation(d.Id(), increaseNodesGroupRequest["NodeGroupId"].(string), resp.(string), 2*Timeout5Minute, &wg, &cm)
		}

		for _, decreaseNodesGroupRequest := range decreaseNodesGroups {
			waitFlag = true
			action := "DecreaseNodes"
			wait := incrementalWait(3*time.Second, 5*time.Second)
			err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
				response, err = client.RpcPost("Emr", "2021-03-20", action, nil, decreaseNodesGroupRequest, false)
				if err != nil {
					if NeedRetry(err) {
						wait()
						return resource.RetryableError(err)
					}
					return resource.NonRetryableError(err)
				}
				return nil
			})
			addDebug(action, response, decreaseNodesGroupRequest)
			if err != nil {
				return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
			}
			resp, err := jsonpath.Get("$.OperationId", response)
			if err != nil {
				return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.OperationId", response)
			}
			wg.Add(1)
			go emrService.WaitForEmrV2Operation(d.Id(), decreaseNodesGroupRequest["NodeGroupId"].(string), resp.(string), 2*Timeout5Minute, &wg, &cm)
		}
		if waitFlag {
			wg.Wait()
			var failedNodeGroupId []string
			cm.Range(func(k, v interface{}) bool {
				if v != nil && v.(bool) == false {
					failedNodeGroupId = append(failedNodeGroupId, k.(string))
				}
				return true
			})
			if len(failedNodeGroupId) > 0 {
				return WrapError(Error("EMR cluster resize found error result with the failed nodeGroupIds: [%s].", strings.Join(failedNodeGroupId, ",")))
			}
		}

		for _, deleteNodeGroupRequest := range deleteNodeGroups {
			action := "DeleteNodeGroup"
			wait := incrementalWait(3*time.Second, 5*time.Second)
			err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
				response, err = client.RpcPost("Emr", "2021-03-20", action, nil, deleteNodeGroupRequest, false)
				if err != nil {
					if NeedRetry(err) {
						wait()
						return resource.RetryableError(err)
					}
					return resource.NonRetryableError(err)
				}
				return nil
			})
			addDebug(action, response, deleteNodeGroupRequest)
			if err != nil {
				return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
			}
		}

		d.SetPartial("node_groups")
	}

	if d.HasChange("bootstrap_scripts") {
		remoteBsMap := map[string]map[string]interface{}{}
		oldBsMap := map[string]map[string]interface{}{}
		newBsMap := map[string]map[string]interface{}{}
		listScriptsRequest := map[string]interface{}{
			"ClusterId":  d.Id(),
			"RegionId":   client.RegionId,
			"ScriptType": "BOOTSTRAP",
			"MaxResults": PageSizeLarge,
		}
		action := "ListScripts"
		wait := incrementalWait(3*time.Second, 5*time.Second)
		err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
			response, err = client.RpcPost("Emr", "2021-03-20", action, nil, listScriptsRequest, false)
			if err != nil {
				if NeedRetry(err) {
					wait()
					return resource.RetryableError(err)
				}
				return resource.NonRetryableError(err)
			}
			return nil
		})
		addDebug(action, response, listScriptsRequest)
		if err != nil {
			return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
		}
		resp, err := jsonpath.Get("$.Scripts", response)

		if err != nil {
			return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.Scripts", response)
		}
		if resp != nil && len(resp.([]interface{})) > 0 {
			for _, v := range resp.([]interface{}) {
				bsMap := v.(map[string]interface{})
				remoteBsMap[bsMap["ScriptName"].(string)] = bsMap
			}
		}

		var toBeCreatedScripts []map[string]interface{}
		var toBeUpdatedScripts []map[string]interface{}
		var toBeDeletedScripts []string
		oldBss, newBss := d.GetChange("bootstrap_scripts")
		if newBss != nil && len(newBss.([]interface{})) > 0 {
			for _, newBsItem := range newBss.([]interface{}) {
				newBs := newBsItem.(map[string]interface{})
				newBsMap[newBs["script_name"].(string)] = newBs
			}
		}
		if oldBss != nil && len(oldBss.([]interface{})) > 0 {
			for _, oldBsItem := range oldBss.([]interface{}) {
				oldBs := oldBsItem.(map[string]interface{})
				oldBsMap[oldBs["script_name"].(string)] = oldBs
			}
		}

		if newBss != nil && len(newBss.([]interface{})) > 0 {
			for _, newBsItem := range newBss.([]interface{}) {
				newBs := newBsItem.(map[string]interface{})
				if remoteBsValue, exists := remoteBsMap[newBs["script_name"].(string)]; exists {
					if !reflect.DeepEqual(newBs, oldBsMap[newBs["script_name"].(string)]) {
						updatedBs := adaptBootstrapScriptRequest(newBs)
						updatedBs["ScriptId"] = remoteBsValue["ScriptId"]
						toBeUpdatedScripts = append(toBeUpdatedScripts, updatedBs)
					}
				} else {
					toBeCreatedScripts = append(toBeCreatedScripts, adaptBootstrapScriptRequest(newBs))
				}
			}
		}

		if oldBss != nil && len(oldBss.([]interface{})) > 0 {
			for _, oldBsItem := range oldBss.([]interface{}) {
				var scriptId string
				oldBs := oldBsItem.(map[string]interface{})
				if remoteBsValue, ok := remoteBsMap[oldBs["script_name"].(string)]; !ok {
					continue
				} else {
					scriptId = remoteBsValue["ScriptId"].(string)
				}
				if _, exists := newBsMap[oldBs["script_name"].(string)]; !exists {
					toBeDeletedScripts = append(toBeDeletedScripts, scriptId)
				}
			}
		}

		if len(toBeCreatedScripts) > 0 {
			createScriptRequest := map[string]interface{}{
				"ClusterId":  d.Id(),
				"RegionId":   client.RegionId,
				"ScriptType": "BOOTSTRAP",
			}
			createScriptRequest["Scripts"] = toBeCreatedScripts
			action = "CreateScript"
			err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
				response, err = client.RpcPost("Emr", "2021-03-20", action, nil, createScriptRequest, false)
				if err != nil {
					if NeedRetry(err) {
						wait()
						return resource.RetryableError(err)
					}
					return resource.NonRetryableError(err)
				}
				return nil
			})
			addDebug(action, response, createScriptRequest)
			if err != nil {
				return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
			}
		}

		if len(toBeUpdatedScripts) > 0 {
			for _, bs := range toBeUpdatedScripts {
				updateScriptRequest := map[string]interface{}{
					"ClusterId":  d.Id(),
					"RegionId":   client.RegionId,
					"ScriptType": "BOOTSTRAP",
					"ScriptId":   bs["ScriptId"],
				}
				delete(bs, "ScriptId")
				updateScriptRequest["Script"] = convertMapToJsonStringIgnoreError(bs)
				action = "UpdateScript"
				err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
					response, err = client.RpcPost("Emr", "2021-03-20", action, nil, updateScriptRequest, false)
					if err != nil {
						if NeedRetry(err) {
							wait()
							return resource.RetryableError(err)
						}
						return resource.NonRetryableError(err)
					}
					return nil
				})
				addDebug(action, response, updateScriptRequest)
				if err != nil {
					return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
				}
			}
		}

		if len(toBeDeletedScripts) > 0 {
			for _, bsId := range toBeDeletedScripts {
				deleteScriptRequest := map[string]interface{}{
					"ClusterId":  d.Id(),
					"RegionId":   client.RegionId,
					"ScriptType": "BOOTSTRAP",
					"ScriptId":   bsId,
				}
				action = "DeleteScript"
				err = resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
					response, err = client.RpcPost("Emr", "2021-03-20", action, nil, deleteScriptRequest, false)
					if err != nil {
						if NeedRetry(err) {
							wait()
							return resource.RetryableError(err)
						}
						return resource.NonRetryableError(err)
					}
					return nil
				})
				addDebug(action, response, deleteScriptRequest)
				if err != nil {
					return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
				}
			}
		}
	}

	d.Partial(false)

	return nil
}

func resourceAlicloudEmrV2ClusterDelete(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*connectivity.AliyunClient)
	emrService := EmrService{client}
	var response map[string]interface{}
	var err error
	if paymentType, ok := d.GetOk("payment_type"); ok && paymentType.(string) == "Subscription" {
		v, err := emrService.ListEmrV2NodeGroups(d.Id(), []string{})
		if err != nil {
			return WrapError(err)
		}
		if v != nil && len(v) > 0 {
			action := "ListNodes"
			request := map[string]interface{}{
				"ClusterId": d.Id(),
				"RegionId":  client.RegionId,
			}
			wait := incrementalWait(3*time.Second, 5*time.Second)
			for _, item := range v {
				nodeGroupMap := item.(map[string]interface{})
				if value, exists := nodeGroupMap["PaymentType"]; exists && value.(string) == "Subscription" {
					request["MaxResults"] = 100
					request["NodeGroupIds"] = []string{nodeGroupMap["NodeGroupId"].(string)}
					err = resource.Retry(d.Timeout(schema.TimeoutDelete), func() *resource.RetryError {
						response, err = client.RpcPost("Emr", "2021-03-20", action, nil, request, true)
						if err != nil {
							if NeedRetry(err) {
								wait()
								return resource.RetryableError(err)
							}
							return resource.NonRetryableError(err)
						}
						return nil
					})
					addDebug(action, response, request)
					if err != nil {
						return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
					}
					nodes, err := jsonpath.Get("$.Nodes", response)
					if err != nil {
						return WrapErrorf(err, FailedGetAttributeMsg, d.Id(), "$.Nodes", response)
					}
					if nodes != nil && len(nodes.([]interface{})) > 0 {
						if err = deleteSubscriptionInstances(d, meta, nodes.([]interface{})); err != nil {
							return WrapError(err)
						}
					}
				}
			}
		}

		// EMR Cluster maybe auto deleted cause all nodes released.
		emrCluster, err := emrService.GetEmrV2Cluster(d.Id())
		if err != nil {
			if NotFoundError(err) {
				return nil
			}
			return WrapError(err)
		}
		if "TERMINATED" == emrCluster["ClusterState"].(string) {
			return nil
		}
	}

	action := "DeleteCluster"
	request := map[string]interface{}{
		"ClusterId": d.Id(),
		"RegionId":  client.RegionId,
	}
	wait := incrementalWait(3*time.Second, 5*time.Second)
	err = resource.Retry(d.Timeout(schema.TimeoutDelete), func() *resource.RetryError {
		response, err = client.RpcPost("Emr", "2021-03-20", action, nil, request, true)
		if err != nil {
			if NeedRetry(err) || strings.Contains(err.Error(), "cluster exists nonempty pre-paid nodeGroups") {
				wait()
				return resource.RetryableError(err)
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})
	addDebug(action, response, request)
	if err != nil {
		return WrapErrorf(err, DefaultErrorMsg, d.Id(), action, AlibabaCloudSdkGoERROR)
	}

	stateConf := BuildStateConf([]string{"TERMINATING"}, []string{}, d.Timeout(schema.TimeoutDelete), 1*time.Millisecond, emrService.EmrV2ClusterStateRefreshFunc(d.Id(), []string{"TERMINATE_FAILED"}))
	stateConf.PollInterval = 5 * time.Second
	if _, err := stateConf.WaitForState(); err != nil {
		return WrapErrorf(err, IdMsg, d.Id())
	}

	return WrapError(emrService.WaitForEmrV2Cluster(d.Id(), Deleted, DefaultTimeoutMedium))
}

func deleteSubscriptionInstances(d *schema.ResourceData, meta interface{}, instances []interface{}) error {
	client := meta.(*connectivity.AliyunClient)
	ecsService := EcsService{client}
	request := ecs.CreateModifyInstanceChargeTypeRequest()
	var instanceIds []interface{}
	for _, item := range instances {
		instanceIds = append(instanceIds, item.(map[string]interface{})["NodeId"])
	}
	request.InstanceIds = convertListToJsonString(instanceIds)
	request.AutoPay = requests.NewBoolean(true)
	request.DryRun = requests.NewBoolean(false)
	request.InstanceChargeType = string(PostPaid)
	if err := resource.Retry(d.Timeout(schema.TimeoutUpdate), func() *resource.RetryError {
		raw, err := client.WithEcsClient(func(ecsClient *ecs.Client) (interface{}, error) {
			return ecsClient.ModifyInstanceChargeType(request)
		})
		if err != nil {
			if NeedRetry(err) || IsExpectedErrors(err, []string{"InternalError"}) {
				time.Sleep(3 * time.Second)
				return resource.RetryableError(err)
			}
			return resource.NonRetryableError(err)
		}
		addDebug(request.GetActionName(), raw, request.RpcRequest, request)
		return nil
	}); err != nil {
		return WrapErrorf(err, DefaultErrorMsg, d.Id(), request.GetActionName(), AlibabaCloudSdkGoERROR)
	}

	for _, instanceId := range instanceIds {
		deleteRequest := ecs.CreateDeleteInstanceRequest()
		deleteRequest.InstanceId = instanceId.(string)
		deleteRequest.Force = requests.NewBoolean(true)

		wait := incrementalWait(1*time.Second, 1*time.Second)
		err := resource.Retry(d.Timeout(schema.TimeoutDelete), func() *resource.RetryError {
			raw, err := client.WithEcsClient(func(ecsClient *ecs.Client) (interface{}, error) {
				return ecsClient.DeleteInstance(deleteRequest)
			})
			if err != nil {
				if IsExpectedErrors(err, []string{"IncorrectInstanceStatus", "DependencyViolation.RouteEntry", "IncorrectInstanceStatus.Initializing"}) {
					return resource.RetryableError(err)
				}
				if IsExpectedErrors(err, []string{Throttling, "LastTokenProcessing"}) {
					wait()
					return resource.RetryableError(err)
				}
				return resource.NonRetryableError(err)
			}
			addDebug(deleteRequest.GetActionName(), raw)
			return nil
		})
		if err != nil {
			if IsExpectedErrors(err, EcsNotFound) {
				return nil
			}
			return WrapErrorf(err, DefaultErrorMsg, d.Id(), deleteRequest.GetActionName(), AlibabaCloudSdkGoERROR)
		}

		stateConf := BuildStateConf([]string{"Pending", "Running", "Stopped", "Stopping"}, []string{}, d.Timeout(schema.TimeoutDelete), 10*time.Second, ecsService.InstanceStateRefreshFunc(d.Id(), []string{}))

		if _, err = stateConf.WaitForState(); err != nil {
			return WrapErrorf(err, IdMsg, d.Id())
		}
	}
	return nil
}

func adaptAutoScalingPolicyRequest(r map[string]interface{}) map[string]interface{} {
	scalingPolicy := map[string]interface{}{}
	if value, exists := r["constraints"]; exists {
		constraints := value.([]interface{})
		if len(constraints) == 1 {
			constraint := map[string]interface{}{}
			constraintMap := constraints[0].(map[string]interface{})
			if constraintValue, constraintExists := constraintMap["max_capacity"]; constraintExists {
				constraint["maxCapacity"] = constraintValue
				constraint["MaxCapacity"] = constraintValue
			}
			if constraintValue, constraintExists := constraintMap["min_capacity"]; constraintExists {
				constraint["minCapacity"] = constraintValue
				constraint["MinCapacity"] = constraintValue
			}
			scalingPolicy["constraints"] = constraint
			scalingPolicy["Constraints"] = constraint
		}
	}
	if value, exists := r["scaling_rules"]; exists {
		var scalingRules []map[string]interface{}
		for _, sr := range value.([]interface{}) {
			scalingRule := map[string]interface{}{}
			scalingRuleMap := sr.(map[string]interface{})
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["rule_name"]; scalingRuleExists {
				scalingRule["RuleName"] = scalingRuleValue
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["trigger_type"]; scalingRuleExists {
				scalingRule["TriggerType"] = scalingRuleValue
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["activity_type"]; scalingRuleExists {
				scalingRule["ActivityType"] = scalingRuleValue
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["adjustment_type"]; scalingRuleExists {
				scalingRule["AdjustmentType"] = scalingRuleValue
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["adjustment_value"]; scalingRuleExists {
				scalingRule["AdjustmentValue"] = scalingRuleValue
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["min_adjustment_value"]; scalingRuleExists {
				scalingRule["MinAdjustmentValue"] = scalingRuleValue
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["time_trigger"]; scalingRuleExists {
				timeTriggers := scalingRuleValue.([]interface{})
				if len(timeTriggers) == 1 {
					timeTrigger := map[string]interface{}{}
					timeTriggerMap := timeTriggers[0].(map[string]interface{})
					if timeTriggerValue, timeTriggerExists := timeTriggerMap["launch_time"]; timeTriggerExists {
						timeTrigger["LaunchTime"] = timeTriggerValue
					}
					if timeTriggerValue, timeTriggerExists := timeTriggerMap["start_time"]; timeTriggerExists {
						timeTrigger["StartTime"] = timeTriggerValue
					}
					if timeTriggerValue, timeTriggerExists := timeTriggerMap["end_time"]; timeTriggerExists {
						timeTrigger["EndTime"] = timeTriggerValue
					}
					if timeTriggerValue, timeTriggerExists := timeTriggerMap["launch_expiration_time"]; timeTriggerExists {
						timeTrigger["LaunchExpirationTime"] = timeTriggerValue
					}
					if timeTriggerValue, timeTriggerExists := timeTriggerMap["recurrence_type"]; timeTriggerExists {
						timeTrigger["RecurrenceType"] = timeTriggerValue
					}
					if timeTriggerValue, timeTriggerExists := timeTriggerMap["recurrence_value"]; timeTriggerExists {
						timeTrigger["RecurrenceValue"] = timeTriggerValue
					}
					scalingRule["TimeTrigger"] = timeTrigger
				}
			}
			if scalingRuleValue, scalingRuleExists := scalingRuleMap["metrics_trigger"]; scalingRuleExists {
				metricsTriggers := scalingRuleValue.([]interface{})
				if len(metricsTriggers) == 1 {
					metricsTrigger := map[string]interface{}{}
					metricsTriggerMap := metricsTriggers[0].(map[string]interface{})
					if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["time_window"]; metricsTriggerExists {
						metricsTrigger["TimeWindow"] = metricsTriggerValue
					}
					if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["evaluation_count"]; metricsTriggerExists {
						metricsTrigger["EvaluationCount"] = metricsTriggerValue
					}
					if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["cool_down_interval"]; metricsTriggerExists {
						metricsTrigger["CoolDownInterval"] = metricsTriggerValue
					}
					if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["condition_logic_operator"]; metricsTriggerExists {
						metricsTrigger["ConditionLogicOperator"] = metricsTriggerValue
					}
					if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["time_constraints"]; metricsTriggerExists {
						var timeConstraints []map[string]interface{}
						for _, tc := range metricsTriggerValue.([]interface{}) {
							timeConstraint := map[string]interface{}{}
							timeConstraintMap := tc.(map[string]interface{})
							if timeConstraintValue, timeConstraintExists := timeConstraintMap["start_time"]; timeConstraintExists {
								timeConstraint["StartTime"] = timeConstraintValue
							}
							if timeConstraintValue, timeConstraintExists := timeConstraintMap["end_time"]; timeConstraintExists {
								timeConstraint["EndTime"] = timeConstraintValue
							}
							timeConstraints = append(timeConstraints, timeConstraint)
						}
						metricsTrigger["TimeConstraints"] = timeConstraints
					}
					if metricsTriggerValue, metricsTriggerExists := metricsTriggerMap["conditions"]; metricsTriggerExists {
						var conditions []map[string]interface{}
						for _, cd := range metricsTriggerValue.([]interface{}) {
							condition := map[string]interface{}{}
							conditionMap := cd.(map[string]interface{})
							if conditionValue, conditionExists := conditionMap["metric_name"]; conditionExists {
								condition["MetricName"] = conditionValue
							}
							if conditionValue, conditionExists := conditionMap["statistics"]; conditionExists {
								condition["Statistics"] = conditionValue
							}
							if conditionValue, conditionExists := conditionMap["comparison_operator"]; conditionExists {
								condition["ComparisonOperator"] = conditionValue
							}
							if conditionValue, conditionExists := conditionMap["threshold"]; conditionExists {
								condition["Threshold"] = conditionValue
							}
							if conditionValue, conditionExists := conditionMap["tags"]; conditionExists {
								var tags []map[string]interface{}
								for _, t := range conditionValue.([]interface{}) {
									tag := map[string]interface{}{}
									tagMap := t.(map[string]interface{})
									if tagValue, tagExists := tagMap["key"]; tagExists {
										tag["Key"] = tagValue
									}
									if tagValue, tagExists := tagMap["value"]; tagExists {
										tag["Value"] = tagValue
									}
									tags = append(tags, tag)
								}
								condition["Tags"] = tags
							}
							conditions = append(conditions, condition)
						}
						metricsTrigger["Conditions"] = conditions
					}
					scalingRule["MetricsTrigger"] = metricsTrigger
				}
			}
			scalingRules = append(scalingRules, scalingRule)
		}
		scalingPolicy["scalingRules"] = scalingRules
		scalingPolicy["ScalingRules"] = scalingRules
	}
	return scalingPolicy
}

func adaptAckConfigRequest(r map[string]interface{}) map[string]interface{} {
	ackConfig := map[string]interface{}{}
	if value, exists := r["ack_instance_id"]; exists {
		ackConfig["AckInstanceId"] = value
	}
	if value, exists := r["node_selectors"]; exists {
		nodeSelectors := value.(*schema.Set).List()
		if len(nodeSelectors) > 0 {
			var nodeSelectorsReq []map[string]interface{}
			for _, ns := range nodeSelectors {
				nodeSelectorsReq = append(nodeSelectorsReq, map[string]interface{}{
					"Key":   ns.(map[string]interface{})["key"],
					"Value": ns.(map[string]interface{})["value"],
				})
			}
			ackConfig["NodeSelectors"] = nodeSelectorsReq
		}
	}
	if value, exists := r["tolerations"]; exists {
		tolerations := value.([]interface{})
		if len(tolerations) > 0 {
			var tolerationsReq []map[string]interface{}
			for _, t := range tolerations {
				toleration := t.(map[string]interface{})
				tolerationsReq = append(tolerationsReq, map[string]interface{}{
					"Key":      toleration["key"],
					"Value":    toleration["value"],
					"Operator": toleration["operator"],
					"Effect":   toleration["effect"],
				})
			}
			ackConfig["Tolerations"] = tolerationsReq
		}
	}
	if value, exists := r["namespace"]; exists {
		ackConfig["Namespace"] = value
	}
	if value, exists := r["request_cpu"]; exists {
		ackConfig["RequestCpu"] = value
	}
	if value, exists := r["request_memory"]; exists {
		ackConfig["RequestMemory"] = value
	}
	if value, exists := r["limit_cpu"]; exists {
		ackConfig["LimitCpu"] = value
	}
	if value, exists := r["limit_memory"]; exists {
		ackConfig["LimitMemory"] = value
	}
	if value, exists := r["custom_labels"]; exists {
		customLabels := value.(*schema.Set).List()
		if len(customLabels) > 0 {
			var customLabelsReq []map[string]interface{}
			for _, cl := range customLabels {
				customLabel := cl.(map[string]interface{})
				customLabelsReq = append(customLabelsReq, map[string]interface{}{
					"Key":   customLabel["key"],
					"Value": customLabel["value"],
				})
			}
			ackConfig["CustomLabels"] = customLabelsReq
		}
	}
	if value, exists := r["custom_annotations"]; exists {
		customAnnotations := value.(*schema.Set).List()
		if len(customAnnotations) > 0 {
			var customAnnotationsReq []map[string]interface{}
			for _, ca := range customAnnotations {
				customAnnotation := ca.(map[string]interface{})
				customAnnotationsReq = append(customAnnotationsReq, map[string]interface{}{
					"Key":   customAnnotation["key"],
					"Value": customAnnotation["value"],
				})
			}
			ackConfig["CustomAnnotations"] = customAnnotationsReq
		}
	}
	if value, exists := r["pvcs"]; exists {
		pvcs := value.([]interface{})
		if len(pvcs) > 0 {
			var pvcsReq []map[string]interface{}
			for _, pvc := range pvcs {
				pvcMap := pvc.(map[string]interface{})
				pvcsReq = append(pvcsReq, map[string]interface{}{
					"DataDiskStorageClass": pvcMap["data_disk_storage_class"],
					"DataDiskSize":         pvcMap["data_disk_size"],
					"Path":                 pvcMap["path"],
					"Name":                 pvcMap["name"],
				})
			}
			ackConfig["Pvcs"] = pvcsReq
		}
	}
	if value, exists := r["volumes"]; exists {
		volumes := value.([]interface{})
		if len(volumes) > 0 {
			var volumesReq []map[string]interface{}
			for _, vl := range volumes {
				volume := vl.(map[string]interface{})
				volumesReq = append(volumesReq, map[string]interface{}{
					"Name": volume["name"],
					"Path": volume["path"],
					"Type": volume["type"],
				})
			}
			ackConfig["Volumes"] = volumesReq
		}
	}
	if value, exists := r["volume_mounts"]; exists {
		volumeMounts := value.([]interface{})
		if len(volumeMounts) > 0 {
			var volumeMountsReq []map[string]interface{}
			for _, vm := range volumeMounts {
				volumeMount := vm.(map[string]interface{})
				volumeMountsReq = append(volumeMountsReq, map[string]interface{}{
					"Name": volumeMount["name"],
					"Path": volumeMount["path"],
				})
			}
			ackConfig["VolumeMounts"] = volumeMountsReq
		}
	}
	if value, exists := r["pre_start_command"]; exists {
		preStartCommand := value.([]interface{})
		if len(preStartCommand) > 0 {
			var preStartCommandReq []string
			for _, prc := range preStartCommand {
				preStartCommandReq = append(preStartCommandReq, prc.(string))
			}
			ackConfig["PreStartCommand"] = preStartCommandReq
		}
	}
	if value, exists := r["pod_affinity"]; exists {
		ackConfig["PodAffinity"] = value
	}
	if value, exists := r["pod_anti_affinity"]; exists {
		ackConfig["PodAntiAffinity"] = value
	}
	if value, exists := r["node_affinity"]; exists {
		ackConfig["NodeAffinity"] = value
	}

	return ackConfig
}

func adaptBootstrapScriptRequest(r map[string]interface{}) map[string]interface{} {
	bsReq := map[string]interface{}{
		"ScriptName":            r["script_name"],
		"ScriptPath":            r["script_path"],
		"ScriptArgs":            r["script_args"],
		"ExecutionMoment":       r["execution_moment"],
		"ExecutionFailStrategy": r["execution_fail_strategy"],
	}
	if value, exists := r["priority"]; exists && value.(int) > 0 {
		bsReq["Priority"] = value.(int)
	}
	if value, exists := r["node_selector"]; exists && len(value.(*schema.Set).List()) == 1 {
		nodeSelectorMap := value.(*schema.Set).List()[0].(map[string]interface{})
		nodeSelector := map[string]interface{}{
			"NodeSelectType": nodeSelectorMap["node_select_type"],
			"NodeGroupId":    nodeSelectorMap["node_group_id"],
			"NodeGroupName":  nodeSelectorMap["node_group_name"],
		}
		if v, ok := nodeSelectorMap["node_names"]; ok && len(v.([]interface{})) > 0 {
			var nodeNames []string
			for _, nn := range v.([]interface{}) {
				nodeNames = append(nodeNames, nn.(string))
			}
			nodeSelector["NodeNames"] = nodeNames
		}
		if v, ok := nodeSelectorMap["node_group_ids"]; ok && len(v.([]interface{})) > 0 {
			var nodeGroupIds []string
			for _, ngId := range v.([]interface{}) {
				nodeGroupIds = append(nodeGroupIds, ngId.(string))
			}
			nodeSelector["NodeGroupIds"] = nodeGroupIds
		}
		if v, ok := nodeSelectorMap["node_group_types"]; ok && len(v.([]interface{})) > 0 {
			var nodeGroupTypes []string
			for _, ngType := range v.([]interface{}) {
				nodeGroupTypes = append(nodeGroupTypes, ngType.(string))
			}
			nodeSelector["NodeGroupTypes"] = nodeGroupTypes
		}
		if v, ok := nodeSelectorMap["node_group_names"]; ok && len(v.([]interface{})) > 0 {
			var nodeGroupNames []string
			for _, ngName := range v.([]interface{}) {
				nodeGroupNames = append(nodeGroupNames, ngName.(string))
			}
			nodeSelector["NodeGroupNames"] = nodeGroupNames
		}
		bsReq["NodeSelector"] = nodeSelector
	}

	return bsReq
}
