diff --git a/.changelog/22096.txt b/.changelog/22096.txt
new file mode 100644
index 00000000000..d7690f8c080
--- /dev/null
+++ b/.changelog/22096.txt
@@ -0,0 +1,3 @@
+```release-note:new-resource
+aws_ecs_task_set
+```
\ No newline at end of file
diff --git a/internal/provider/provider.go b/internal/provider/provider.go
index cdbc15045d3..3edae25a9a7 100644
--- a/internal/provider/provider.go
+++ b/internal/provider/provider.go
@@ -1130,6 +1130,7 @@ func Provider() *schema.Provider {
 			"aws_ecs_service":                 ecs.ResourceService(),
 			"aws_ecs_tag":                     ecs.ResourceTag(),
 			"aws_ecs_task_definition":         ecs.ResourceTaskDefinition(),
+			"aws_ecs_task_set":                ecs.ResourceTaskSet(),
 
 			"aws_efs_access_point":       efs.ResourceAccessPoint(),
 			"aws_efs_backup_policy":      efs.ResourceBackupPolicy(),
diff --git a/internal/service/ecs/flex.go b/internal/service/ecs/flex.go
index 68f5ca395c7..1bec5029000 100644
--- a/internal/service/ecs/flex.go
+++ b/internal/service/ecs/flex.go
@@ -54,3 +54,129 @@ func flattenECSLoadBalancers(list []*ecs.LoadBalancer) []map[string]interface{}
 	}
 	return result
 }
+
+// Expand for an array of load balancers and
+// returns ecs.LoadBalancer compatible objects for an ECS TaskSet
+func expandTaskSetLoadBalancers(l []interface{}) []*ecs.LoadBalancer {
+	if len(l) == 0 || l[0] == nil {
+		return nil
+	}
+
+	loadBalancers := make([]*ecs.LoadBalancer, 0, len(l))
+
+	// Loop over our configured load balancers and create
+	// an array of aws-sdk-go compatible objects
+	for _, lRaw := range l {
+		data := lRaw.(map[string]interface{})
+
+		l := &ecs.LoadBalancer{}
+
+		if v, ok := data["container_name"].(string); ok && v != "" {
+			l.ContainerName = aws.String(v)
+		}
+
+		if v, ok := data["container_port"].(int); ok {
+			l.ContainerPort = aws.Int64(int64(v))
+		}
+
+		if v, ok := data["load_balancer_name"]; ok && v.(string) != "" {
+			l.LoadBalancerName = aws.String(v.(string))
+		}
+		if v, ok := data["target_group_arn"]; ok && v.(string) != "" {
+			l.TargetGroupArn = aws.String(v.(string))
+		}
+
+		loadBalancers = append(loadBalancers, l)
+	}
+
+	return loadBalancers
+}
+
+// Flattens an array of ECS LoadBalancers (of an ECS TaskSet) into a []map[string]interface{}
+func flattenTaskSetLoadBalancers(list []*ecs.LoadBalancer) []map[string]interface{} {
+	result := make([]map[string]interface{}, 0, len(list))
+	for _, loadBalancer := range list {
+		l := map[string]interface{}{
+			"container_name": loadBalancer.ContainerName,
+			"container_port": loadBalancer.ContainerPort,
+		}
+
+		if loadBalancer.LoadBalancerName != nil {
+			l["load_balancer_name"] = loadBalancer.LoadBalancerName
+		}
+
+		if loadBalancer.TargetGroupArn != nil {
+			l["target_group_arn"] = loadBalancer.TargetGroupArn
+		}
+
+		result = append(result, l)
+	}
+	return result
+}
+
+// Expand for an array of service registries and
+// returns ecs.ServiceRegistry compatible objects for an ECS TaskSet
+func expandServiceRegistries(l []interface{}) []*ecs.ServiceRegistry {
+	if len(l) == 0 || l[0] == nil {
+		return nil
+	}
+
+	result := make([]*ecs.ServiceRegistry, 0, len(l))
+
+	for _, v := range l {
+		m := v.(map[string]interface{})
+		sr := &ecs.ServiceRegistry{
+			RegistryArn: aws.String(m["registry_arn"].(string)),
+		}
+		if raw, ok := m["container_name"].(string); ok && raw != "" {
+			sr.ContainerName = aws.String(raw)
+		}
+		if raw, ok := m["container_port"].(int); ok && raw > 0 {
+			sr.ContainerPort = aws.Int64(int64(raw))
+		}
+		if raw, ok := m["port"].(int); ok && raw > 0 {
+			sr.Port = aws.Int64(int64(raw))
+		}
+		result = append(result, sr)
+	}
+
+	return result
+}
+
+// Expand for an array of scale configurations and
+// returns an ecs.Scale compatible object for an ECS TaskSet
+func expandScale(l []interface{}) *ecs.Scale {
+	if len(l) == 0 || l[0] == nil {
+		return nil
+	}
+
+	tfMap, ok := l[0].(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	result := &ecs.Scale{}
+
+	if v, ok := tfMap["unit"].(string); ok && v != "" {
+		result.Unit = aws.String(v)
+	}
+
+	if v, ok := tfMap["value"].(float64); ok {
+		result.Value = aws.Float64(v)
+	}
+
+	return result
+}
+
+// Flattens an ECS Scale configuration into a []map[string]interface{}
+func flattenScale(scale *ecs.Scale) []map[string]interface{} {
+	if scale == nil {
+		return nil
+	}
+
+	m := make(map[string]interface{})
+	m["unit"] = aws.StringValue(scale.Unit)
+	m["value"] = aws.Float64Value(scale.Value)
+
+	return []map[string]interface{}{m}
+}
diff --git a/internal/service/ecs/status.go b/internal/service/ecs/status.go
index c16da4ec6d3..fce46b8d2e3 100644
--- a/internal/service/ecs/status.go
+++ b/internal/service/ecs/status.go
@@ -21,6 +21,10 @@ const (
 
 	clusterStatusError = "ERROR"
 	clusterStatusNone  = "NONE"
+
+	taskSetStatusActive   = "ACTIVE"
+	taskSetStatusDraining = "DRAINING"
+	taskSetStatusPrimary  = "PRIMARY"
 )
 
 func statusCapacityProvider(conn *ecs.ECS, arn string) resource.StateRefreshFunc {
@@ -100,3 +104,47 @@ func statusCluster(conn *ecs.ECS, arn string) resource.StateRefreshFunc {
 		return output, aws.StringValue(output.Clusters[0].Status), err
 	}
 }
+
+func stabilityStatusTaskSet(conn *ecs.ECS, taskSetID, service, cluster string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		input := &ecs.DescribeTaskSetsInput{
+			Cluster:  aws.String(cluster),
+			Service:  aws.String(service),
+			TaskSets: aws.StringSlice([]string{taskSetID}),
+		}
+
+		output, err := conn.DescribeTaskSets(input)
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		if output == nil || len(output.TaskSets) == 0 {
+			return nil, "", nil
+		}
+
+		return output.TaskSets[0], aws.StringValue(output.TaskSets[0].StabilityStatus), nil
+	}
+}
+
+func statusTaskSet(conn *ecs.ECS, taskSetID, service, cluster string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		input := &ecs.DescribeTaskSetsInput{
+			Cluster:  aws.String(cluster),
+			Service:  aws.String(service),
+			TaskSets: aws.StringSlice([]string{taskSetID}),
+		}
+
+		output, err := conn.DescribeTaskSets(input)
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		if output == nil || len(output.TaskSets) == 0 {
+			return nil, "", nil
+		}
+
+		return output.TaskSets[0], aws.StringValue(output.TaskSets[0].Status), nil
+	}
+}
diff --git a/internal/service/ecs/task_set.go b/internal/service/ecs/task_set.go
new file mode 100644
index 00000000000..1165eb56fce
--- /dev/null
+++ b/internal/service/ecs/task_set.go
@@ -0,0 +1,545 @@
+package ecs
+
+import (
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
+	"github.com/hashicorp/terraform-provider-aws/internal/conns"
+	tfiam "github.com/hashicorp/terraform-provider-aws/internal/service/iam"
+	tftags "github.com/hashicorp/terraform-provider-aws/internal/tags"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
+	"github.com/hashicorp/terraform-provider-aws/internal/verify"
+)
+
+func ResourceTaskSet() *schema.Resource {
+	return &schema.Resource{
+		Create: resourceTaskSetCreate,
+		Read:   resourceTaskSetRead,
+		Update: resourceTaskSetUpdate,
+		Delete: resourceTaskSetDelete,
+		Importer: &schema.ResourceImporter{
+			State: schema.ImportStatePassthrough,
+		},
+
+		Schema: map[string]*schema.Schema{
+			"arn": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"service": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"cluster": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"external_id": {
+				Type:     schema.TypeString,
+				Optional: true,
+				ForceNew: true,
+				Computed: true,
+			},
+
+			"task_definition": {
+				Type:     schema.TypeString,
+				Required: true,
+				ForceNew: true,
+			},
+
+			"task_set_id": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"network_configuration": {
+				Type:     schema.TypeList,
+				MaxItems: 1,
+				Optional: true,
+				ForceNew: true,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"security_groups": {
+							Type:     schema.TypeSet,
+							MaxItems: 5,
+							Optional: true,
+							ForceNew: true,
+							Elem:     &schema.Schema{Type: schema.TypeString},
+						},
+						"subnets": {
+							Type:     schema.TypeSet,
+							MaxItems: 16,
+							Required: true,
+							ForceNew: true,
+							Elem:     &schema.Schema{Type: schema.TypeString},
+						},
+						"assign_public_ip": {
+							Type:     schema.TypeBool,
+							Optional: true,
+							Default:  false,
+							ForceNew: true,
+						},
+					},
+				},
+			},
+
+			// If you are using the CodeDeploy or an external deployment controller,
+			// multiple target groups are not supported.
+			// https://docs.aws.amazon.com/AmazonECS/latest/developerguide/register-multiple-targetgroups.html
+			"load_balancer": {
+				Type:     schema.TypeSet,
+				Optional: true,
+				ForceNew: true,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"load_balancer_name": {
+							Type:     schema.TypeString,
+							Optional: true,
+							ForceNew: true,
+						},
+						"target_group_arn": {
+							Type:         schema.TypeString,
+							Optional:     true,
+							ForceNew:     true,
+							ValidateFunc: verify.ValidARN,
+						},
+						"container_name": {
+							Type:     schema.TypeString,
+							Required: true,
+							ForceNew: true,
+						},
+						"container_port": {
+							Type:         schema.TypeInt,
+							Optional:     true,
+							ForceNew:     true,
+							ValidateFunc: validation.IsPortNumber,
+						},
+					},
+				},
+			},
+
+			"service_registries": {
+				Type:     schema.TypeList,
+				MaxItems: 1,
+				Optional: true,
+				ForceNew: true,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"container_name": {
+							Type:     schema.TypeString,
+							Optional: true,
+							ForceNew: true,
+						},
+						"container_port": {
+							Type:         schema.TypeInt,
+							Optional:     true,
+							ForceNew:     true,
+							ValidateFunc: validation.IsPortNumber,
+						},
+						"port": {
+							Type:         schema.TypeInt,
+							Optional:     true,
+							ForceNew:     true,
+							ValidateFunc: validation.IsPortNumber,
+						},
+						"registry_arn": {
+							Type:         schema.TypeString,
+							Required:     true,
+							ForceNew:     true,
+							ValidateFunc: verify.ValidARN,
+						},
+					},
+				},
+			},
+
+			"launch_type": {
+				Type:          schema.TypeString,
+				Optional:      true,
+				ForceNew:      true,
+				Computed:      true,
+				ValidateFunc:  validation.StringInSlice(ecs.LaunchType_Values(), false),
+				ConflictsWith: []string{"capacity_provider_strategy"},
+			},
+
+			"capacity_provider_strategy": {
+				Type:          schema.TypeSet,
+				Optional:      true,
+				ForceNew:      true,
+				ConflictsWith: []string{"launch_type"},
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"base": {
+							Type:         schema.TypeInt,
+							Optional:     true,
+							ValidateFunc: validation.IntBetween(0, 100000),
+							ForceNew:     true,
+						},
+
+						"capacity_provider": {
+							Type:     schema.TypeString,
+							Required: true,
+							ForceNew: true,
+						},
+
+						"weight": {
+							Type:         schema.TypeInt,
+							Required:     true,
+							ValidateFunc: validation.IntBetween(0, 1000),
+							ForceNew:     true,
+						},
+					},
+				},
+			},
+
+			"platform_version": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
+			},
+
+			"scale": {
+				Type:     schema.TypeList,
+				MaxItems: 1,
+				Optional: true,
+				Computed: true,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"unit": {
+							Type:         schema.TypeString,
+							Optional:     true,
+							Default:      ecs.ScaleUnitPercent,
+							ValidateFunc: validation.StringInSlice(ecs.ScaleUnit_Values(), false),
+						},
+						"value": {
+							Type:         schema.TypeFloat,
+							Optional:     true,
+							ValidateFunc: validation.FloatBetween(0.0, 100.0),
+						},
+					},
+				},
+			},
+
+			"force_delete": {
+				Type:     schema.TypeBool,
+				Optional: true,
+			},
+
+			"stability_status": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"status": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+
+			"tags": tftags.TagsSchema(),
+
+			"tags_all": tftags.TagsSchemaComputed(),
+
+			"wait_until_stable": {
+				Type:     schema.TypeBool,
+				Optional: true,
+				Default:  false,
+			},
+
+			"wait_until_stable_timeout": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Default:  "10m",
+				ValidateFunc: func(v interface{}, k string) (ws []string, errors []error) {
+					value := v.(string)
+					duration, err := time.ParseDuration(value)
+					if err != nil {
+						errors = append(errors, fmt.Errorf(
+							"%q cannot be parsed as a duration: %w", k, err))
+					}
+					if duration < 0 {
+						errors = append(errors, fmt.Errorf(
+							"%q must be greater than zero", k))
+					}
+					return
+				},
+			},
+		},
+
+		CustomizeDiff: verify.SetTagsDiff,
+	}
+}
+
+func resourceTaskSetCreate(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*conns.AWSClient).ECSConn
+	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
+	tags := defaultTagsConfig.MergeTags(tftags.New(d.Get("tags").(map[string]interface{})))
+
+	cluster := d.Get("cluster").(string)
+	service := d.Get("service").(string)
+	input := &ecs.CreateTaskSetInput{
+		ClientToken:    aws.String(resource.UniqueId()),
+		Cluster:        aws.String(cluster),
+		Service:        aws.String(service),
+		TaskDefinition: aws.String(d.Get("task_definition").(string)),
+	}
+
+	if len(tags) > 0 {
+		input.Tags = Tags(tags.IgnoreAWS())
+	}
+
+	if v, ok := d.GetOk("capacity_provider_strategy"); ok && v.(*schema.Set).Len() > 0 {
+		input.CapacityProviderStrategy = expandEcsCapacityProviderStrategy(v.(*schema.Set))
+	}
+
+	if v, ok := d.GetOk("external_id"); ok {
+		input.ExternalId = aws.String(v.(string))
+	}
+
+	if v, ok := d.GetOk("launch_type"); ok {
+		input.LaunchType = aws.String(v.(string))
+	}
+
+	if v, ok := d.GetOk("load_balancer"); ok && v.(*schema.Set).Len() > 0 {
+		input.LoadBalancers = expandTaskSetLoadBalancers(v.(*schema.Set).List())
+	}
+
+	if v, ok := d.GetOk("network_configuration"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
+		input.NetworkConfiguration = expandEcsNetworkConfiguration(v.([]interface{}))
+	}
+
+	if v, ok := d.GetOk("platform_version"); ok {
+		input.PlatformVersion = aws.String(v.(string))
+	}
+
+	if v, ok := d.GetOk("scale"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
+		input.Scale = expandScale(v.([]interface{}))
+	}
+
+	if v, ok := d.GetOk("service_registries"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
+		input.ServiceRegistries = expandServiceRegistries(v.([]interface{}))
+	}
+
+	// Retry due to AWS IAM & ECS eventual consistency
+	output, err := tfresource.RetryWhen(
+		tfiam.PropagationTimeout+taskSetCreateTimeout,
+		func() (interface{}, error) {
+			return conn.CreateTaskSet(input)
+		},
+		func(err error) (bool, error) {
+			if tfawserr.ErrCodeEquals(err, ecs.ErrCodeClusterNotFoundException, ecs.ErrCodeServiceNotFoundException, ecs.ErrCodeTaskSetNotFoundException) ||
+				tfawserr.ErrMessageContains(err, ecs.ErrCodeInvalidParameterException, "does not have an associated load balancer") {
+				return true, err
+			}
+			return false, err
+		},
+	)
+
+	if err != nil {
+		return fmt.Errorf("error creating ECS TaskSet: %w", err)
+	}
+
+	result, ok := output.(*ecs.CreateTaskSetOutput)
+	if !ok || result == nil || result.TaskSet == nil {
+		return fmt.Errorf("error creating ECS TaskSet: empty output")
+	}
+
+	taskSetId := aws.StringValue(result.TaskSet.Id)
+
+	d.SetId(fmt.Sprintf("%s,%s,%s", taskSetId, service, cluster))
+
+	if d.Get("wait_until_stable").(bool) {
+		timeout, _ := time.ParseDuration(d.Get("wait_until_stable_timeout").(string))
+		if err := waitTaskSetStable(conn, timeout, taskSetId, service, cluster); err != nil {
+			return fmt.Errorf("error waiting for ECS TaskSet (%s) to be stable: %w", d.Id(), err)
+		}
+	}
+
+	return resourceTaskSetRead(d, meta)
+}
+
+func resourceTaskSetRead(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*conns.AWSClient).ECSConn
+	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
+	ignoreTagsConfig := meta.(*conns.AWSClient).IgnoreTagsConfig
+
+	taskSetId, service, cluster, err := TaskSetParseID(d.Id())
+
+	if err != nil {
+		return err
+	}
+
+	input := &ecs.DescribeTaskSetsInput{
+		Cluster:  aws.String(cluster),
+		Include:  aws.StringSlice([]string{ecs.TaskSetFieldTags}),
+		Service:  aws.String(service),
+		TaskSets: aws.StringSlice([]string{taskSetId}),
+	}
+
+	out, err := conn.DescribeTaskSets(input)
+
+	if !d.IsNewResource() && tfawserr.ErrCodeEquals(err, ecs.ErrCodeClusterNotFoundException, ecs.ErrCodeServiceNotFoundException, ecs.ErrCodeTaskSetNotFoundException) {
+		log.Printf("[WARN] ECS TaskSet (%s) not found, removing from state", d.Id())
+		d.SetId("")
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("error reading ECS TaskSet (%s): %w", d.Id(), err)
+	}
+
+	if out == nil || len(out.TaskSets) == 0 {
+		if d.IsNewResource() {
+			return fmt.Errorf("error reading ECS TaskSet (%s): empty output after creation", d.Id())
+		}
+		log.Printf("[WARN] ECS TaskSet (%s) not found, removing from state", d.Id())
+		d.SetId("")
+		return nil
+	}
+
+	taskSet := out.TaskSets[0]
+
+	d.Set("arn", taskSet.TaskSetArn)
+	d.Set("cluster", cluster)
+	d.Set("launch_type", taskSet.LaunchType)
+	d.Set("platform_version", taskSet.PlatformVersion)
+	d.Set("external_id", taskSet.ExternalId)
+	d.Set("service", service)
+	d.Set("status", taskSet.Status)
+	d.Set("stability_status", taskSet.StabilityStatus)
+	d.Set("task_definition", taskSet.TaskDefinition)
+	d.Set("task_set_id", taskSet.Id)
+
+	if err := d.Set("capacity_provider_strategy", flattenEcsCapacityProviderStrategy(taskSet.CapacityProviderStrategy)); err != nil {
+		return fmt.Errorf("error setting capacity_provider_strategy: %w", err)
+	}
+
+	if err := d.Set("load_balancer", flattenTaskSetLoadBalancers(taskSet.LoadBalancers)); err != nil {
+		return fmt.Errorf("error setting load_balancer: %w", err)
+	}
+
+	if err := d.Set("network_configuration", flattenEcsNetworkConfiguration(taskSet.NetworkConfiguration)); err != nil {
+		return fmt.Errorf("error setting network_configuration: %w", err)
+	}
+
+	if err := d.Set("scale", flattenScale(taskSet.Scale)); err != nil {
+		return fmt.Errorf("error setting scale: %w", err)
+	}
+
+	if err := d.Set("service_registries", flattenServiceRegistries(taskSet.ServiceRegistries)); err != nil {
+		return fmt.Errorf("error setting service_registries: %w", err)
+	}
+
+	tags := KeyValueTags(taskSet.Tags).IgnoreAWS().IgnoreConfig(ignoreTagsConfig)
+
+	//lintignore:AWSR002
+	if err := d.Set("tags", tags.RemoveDefaultConfig(defaultTagsConfig).Map()); err != nil {
+		return fmt.Errorf("error setting tags: %w", err)
+	}
+
+	if err := d.Set("tags_all", tags.Map()); err != nil {
+		return fmt.Errorf("error setting tags_all: %w", err)
+	}
+
+	return nil
+}
+
+func resourceTaskSetUpdate(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*conns.AWSClient).ECSConn
+
+	if d.HasChangesExcept("tags", "tags_all") {
+		taskSetId, service, cluster, err := TaskSetParseID(d.Id())
+
+		if err != nil {
+			return err
+		}
+
+		input := &ecs.UpdateTaskSetInput{
+			Cluster: aws.String(cluster),
+			Service: aws.String(service),
+			TaskSet: aws.String(taskSetId),
+			Scale:   expandScale(d.Get("scale").([]interface{})),
+		}
+
+		_, err = conn.UpdateTaskSet(input)
+
+		if err != nil {
+			return fmt.Errorf("error updating ECS TaskSet (%s): %w", d.Id(), err)
+		}
+
+		if d.Get("wait_until_stable").(bool) {
+			timeout, _ := time.ParseDuration(d.Get("wait_until_stable_timeout").(string))
+			if err := waitTaskSetStable(conn, timeout, taskSetId, service, cluster); err != nil {
+				return fmt.Errorf("error waiting for ECS TaskSet (%s) to be stable after update: %w", d.Id(), err)
+			}
+		}
+	}
+
+	if d.HasChange("tags_all") {
+		o, n := d.GetChange("tags_all")
+
+		if err := UpdateTags(conn, d.Get("arn").(string), o, n); err != nil {
+			return fmt.Errorf("error updating ECS TaskSet (%s) tags: %w", d.Id(), err)
+		}
+	}
+
+	return resourceTaskSetRead(d, meta)
+}
+
+func resourceTaskSetDelete(d *schema.ResourceData, meta interface{}) error {
+	conn := meta.(*conns.AWSClient).ECSConn
+
+	taskSetId, service, cluster, err := TaskSetParseID(d.Id())
+
+	if err != nil {
+		return err
+	}
+
+	input := &ecs.DeleteTaskSetInput{
+		Cluster: aws.String(cluster),
+		Service: aws.String(service),
+		TaskSet: aws.String(taskSetId),
+		Force:   aws.Bool(d.Get("force_delete").(bool)),
+	}
+
+	_, err = conn.DeleteTaskSet(input)
+
+	if tfawserr.ErrCodeEquals(err, ecs.ErrCodeTaskSetNotFoundException) {
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("error deleting ECS TaskSet (%s): %w", d.Id(), err)
+	}
+
+	if err := waitTaskSetDeleted(conn, taskSetId, service, cluster); err != nil {
+		if tfawserr.ErrCodeEquals(err, ecs.ErrCodeTaskSetNotFoundException) {
+			return nil
+		}
+		return fmt.Errorf("error waiting for ECS TaskSet (%s) to delete: %w", d.Id(), err)
+	}
+
+	return nil
+}
+
+func TaskSetParseID(id string) (string, string, string, error) {
+	parts := strings.Split(id, ",")
+
+	if len(parts) != 3 || parts[0] == "" || parts[1] == "" || parts[2] == "" {
+		return "", "", "", fmt.Errorf("unexpected format of ID (%q), expected TASK_SET_ID,SERVICE,CLUSTER", id)
+	}
+
+	return parts[0], parts[1], parts[2], nil
+}
diff --git a/internal/service/ecs/task_set_test.go b/internal/service/ecs/task_set_test.go
new file mode 100644
index 00000000000..965c8720187
--- /dev/null
+++ b/internal/service/ecs/task_set_test.go
@@ -0,0 +1,1055 @@
+package ecs_test
+
+import (
+	"fmt"
+	"regexp"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ecs"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	sdkacctest "github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
+	"github.com/hashicorp/terraform-provider-aws/internal/acctest"
+	"github.com/hashicorp/terraform-provider-aws/internal/conns"
+	tfecs "github.com/hashicorp/terraform-provider-aws/internal/service/ecs"
+)
+
+func TestAccECSTaskSet_basic(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetBasicConfig(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					acctest.MatchResourceAttrRegionalARN(resourceName, "arn", "ecs", regexp.MustCompile(fmt.Sprintf("task-set/%[1]s/%[1]s/ecs-svc/.+", rName))),
+					resource.TestCheckResourceAttr(resourceName, "service_registries.#", "0"),
+					resource.TestCheckResourceAttr(resourceName, "load_balancer.#", "0"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withExternalId(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithExternalIdConfig(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "service_registries.#", "0"),
+					resource.TestCheckResourceAttr(resourceName, "load_balancer.#", "0"),
+					resource.TestCheckResourceAttr(resourceName, "external_id", "TEST_ID"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withScale(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithScaleConfig(rName, 0.0),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "scale.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "scale.0.unit", ecs.ScaleUnitPercent),
+					resource.TestCheckResourceAttr(resourceName, "scale.0.value", "0"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+			{
+				Config: testAccTaskSetWithScaleConfig(rName, 100.0),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "scale.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "scale.0.unit", ecs.ScaleUnitPercent),
+					resource.TestCheckResourceAttr(resourceName, "scale.0.value", "100"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_disappears(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetBasicConfig(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					acctest.CheckResourceDisappears(acctest.Provider, tfecs.ResourceTaskSet(), resourceName),
+				),
+				ExpectNonEmptyPlan: true,
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withCapacityProviderStrategy(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithCapacityProviderStrategy(rName, 1, 0),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+			{
+				Config: testAccTaskSetWithCapacityProviderStrategy(rName, 10, 1),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withMultipleCapacityProviderStrategies(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithMultipleCapacityProviderStrategies(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "capacity_provider_strategy.#", "2"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withAlb(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithAlb(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "load_balancer.#", "1"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withLaunchTypeFargate(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithLaunchTypeFargate(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "launch_type", "FARGATE"),
+					resource.TestCheckResourceAttr(resourceName, "network_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "network_configuration.0.assign_public_ip", "false"),
+					resource.TestCheckResourceAttr(resourceName, "network_configuration.0.security_groups.#", "2"),
+					resource.TestCheckResourceAttr(resourceName, "network_configuration.0.subnets.#", "2"),
+					resource.TestCheckResourceAttrSet(resourceName, "platform_version"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withLaunchTypeFargateAndPlatformVersion(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetWithLaunchTypeFargateAndPlatformVersion(rName, "1.3.0"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "platform_version", "1.3.0"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+			{
+				Config: testAccTaskSetWithLaunchTypeFargateAndPlatformVersion(rName, "1.4.0"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "platform_version", "1.4.0"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_withServiceRegistries(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSet_withServiceRegistries(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "service_registries.#", "1"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+		},
+	})
+}
+
+func TestAccECSTaskSet_Tags(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_ecs_task_set.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, ecs.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckTaskSetDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTaskSetConfigTags1(rName, "key1", "value1"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key1", "value1"),
+				),
+			},
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateVerifyIgnore: []string{
+					"wait_until_stable",
+					"wait_until_stable_timeout",
+				},
+			},
+			{
+				Config: testAccTaskSetConfigTags2(rName, "key1", "value1updated", "key2", "value2"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "2"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key1", "value1updated"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key2", "value2"),
+				),
+			},
+			{
+				Config: testAccTaskSetConfigTags1(rName, "key2", "value2"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckTaskSetExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key2", "value2"),
+				),
+			},
+		},
+	})
+}
+
+//////////////
+// Fixtures //
+//////////////
+
+func testAccTaskSetBaseConfig(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_ecs_cluster" "test" {
+  name = %[1]q
+}
+
+resource "aws_ecs_task_definition" "test" {
+  family                = %[1]q
+  container_definitions = <<DEFINITION
+[
+  {
+    "cpu": 128,
+    "essential": true,
+    "image": "mongo:latest",
+    "memory": 128,
+    "name": "mongodb"
+  }
+]
+DEFINITION
+}
+
+resource "aws_ecs_service" "test" {
+  name          = %[1]q
+  cluster       = aws_ecs_cluster.test.id
+  desired_count = 1
+  deployment_controller {
+    type = "EXTERNAL"
+  }
+}
+`, rName)
+}
+
+func testAccTaskSetBasicConfig(rName string) string {
+	return acctest.ConfigCompose(
+		testAccTaskSetBaseConfig(rName),
+		`
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+}
+`)
+}
+
+func testAccTaskSetWithExternalIdConfig(rName string) string {
+	return acctest.ConfigCompose(
+		testAccTaskSetBaseConfig(rName),
+		`
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  external_id     = "TEST_ID"
+}
+`)
+}
+
+func testAccTaskSetWithScaleConfig(rName string, scale float64) string {
+	return acctest.ConfigCompose(
+		testAccTaskSetBaseConfig(rName),
+		fmt.Sprintf(`
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  scale {
+    value = %[1]f
+  }
+}
+`, scale))
+}
+
+func testAccTaskSetWithCapacityProviderStrategy(rName string, weight, base int) string {
+	return acctest.ConfigCompose(
+		testAccCapacityProviderBaseConfig(rName),
+		testAccTaskSetBaseConfig(rName),
+		fmt.Sprintf(`
+resource "aws_ecs_capacity_provider" "test" {
+  name = %[1]q
+  auto_scaling_group_provider {
+    auto_scaling_group_arn = aws_autoscaling_group.test.arn
+  }
+}
+
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  capacity_provider_strategy {
+    capacity_provider = aws_ecs_capacity_provider.test.name
+    weight            = %[2]d
+    base              = %[3]d
+  }
+}
+`, rName, weight, base))
+}
+
+func testAccTaskSetWithMultipleCapacityProviderStrategies(rName string) string {
+	return fmt.Sprintf(`
+resource "aws_vpc" "test" {
+  cidr_block = "10.10.0.0/16"
+  tags = {
+    Name = "tf-acc-ecs-service-with-multiple-capacity-providers"
+  }
+}
+
+resource "aws_security_group" "test" {
+  name        = %[1]q
+  description = "Allow all inbound traffic"
+  vpc_id      = aws_vpc.test.id
+  ingress {
+    protocol    = "tcp"
+    from_port   = 80
+    to_port     = 8000
+    cidr_blocks = [aws_vpc.test.cidr_block]
+  }
+}
+
+resource "aws_subnet" "test" {
+  cidr_block = cidrsubnet(aws_vpc.test.cidr_block, 8, 1)
+  vpc_id     = aws_vpc.test.id
+  tags = {
+    Name = "tf-acc-ecs-service-with-multiple-capacity-providers"
+  }
+}
+
+resource "aws_ecs_cluster" "test" {
+  name = %[1]q
+
+  capacity_providers = ["FARGATE_SPOT", "FARGATE"]
+
+  default_capacity_provider_strategy {
+    capacity_provider = "FARGATE_SPOT"
+    weight            = 1
+    base              = 1
+  }
+
+  default_capacity_provider_strategy {
+    capacity_provider = "FARGATE"
+    weight            = 1
+  }
+}
+
+resource "aws_ecs_service" "test" {
+  name          = %[1]q
+  cluster       = aws_ecs_cluster.test.id
+  desired_count = 1
+  deployment_controller {
+    type = "EXTERNAL"
+  }
+}
+
+resource "aws_ecs_task_definition" "test" {
+  family                   = %[1]q
+  network_mode             = "awsvpc"
+  requires_compatibilities = ["FARGATE"]
+  cpu                      = "256"
+  memory                   = "512"
+  container_definitions    = <<DEFINITION
+[
+  {
+    "cpu": 256,
+    "essential": true,
+    "image": "mongo:latest",
+    "memory": 512,
+    "name": "mongodb",
+    "networkMode": "awsvpc"
+  }
+]
+DEFINITION
+}
+
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  network_configuration {
+    security_groups  = [aws_security_group.test.id]
+    subnets          = [aws_subnet.test.id]
+    assign_public_ip = false
+  }
+  capacity_provider_strategy {
+    capacity_provider = "FARGATE"
+    weight            = 1
+  }
+  capacity_provider_strategy {
+    capacity_provider = "FARGATE_SPOT"
+    weight            = 1
+  }
+}
+`, rName)
+}
+
+func testAccTaskSetWithAlb(rName string) string {
+	return acctest.ConfigCompose(
+		acctest.ConfigAvailableAZsNoOptIn(),
+		fmt.Sprintf(`
+resource "aws_vpc" "test" {
+  cidr_block = "10.10.0.0/16"
+  tags = {
+    Name = "terraform-testacc-ecs-service-with-alb"
+  }
+}
+
+resource "aws_subnet" "test" {
+  count             = 2
+  cidr_block        = cidrsubnet(aws_vpc.test.cidr_block, 8, count.index)
+  availability_zone = data.aws_availability_zones.available.names[count.index]
+  vpc_id            = aws_vpc.test.id
+  tags = {
+    Name = "tf-acc-ecs-service-with-alb"
+  }
+}
+
+resource "aws_ecs_cluster" "test" {
+  name = %[1]q
+}
+
+resource "aws_ecs_task_definition" "test" {
+  family                = %[1]q
+  container_definitions = <<DEFINITION
+[
+  {
+    "cpu": 256,
+    "essential": true,
+    "image": "ghost:latest",
+    "memory": 512,
+    "name": "ghost",
+    "portMappings": [
+      {
+        "containerPort": 2368,
+        "hostPort": 8080
+      }
+    ]
+  }
+]
+DEFINITION
+}
+
+resource "aws_lb_target_group" "test" {
+  name     = aws_lb.test.name
+  port     = 80
+  protocol = "HTTP"
+  vpc_id   = aws_vpc.test.id
+}
+
+resource "aws_lb" "test" {
+  name     = %[1]q
+  internal = true
+  subnets  = aws_subnet.test.*.id
+}
+
+resource "aws_lb_listener" "test" {
+  load_balancer_arn = aws_lb.test.id
+  port              = "80"
+  protocol          = "HTTP"
+  default_action {
+    target_group_arn = aws_lb_target_group.test.id
+    type             = "forward"
+  }
+}
+
+resource "aws_ecs_service" "test" {
+  name          = %[1]q
+  cluster       = aws_ecs_cluster.test.id
+  desired_count = 1
+  deployment_controller {
+    type = "EXTERNAL"
+  }
+}
+
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  load_balancer {
+    target_group_arn = aws_lb_target_group.test.id
+    container_name   = "ghost"
+    container_port   = "2368"
+  }
+}
+`, rName))
+}
+
+func testAccTaskSetConfigTags1(rName, tag1Key, tag1Value string) string {
+	return acctest.ConfigCompose(
+		testAccTaskSetBaseConfig(rName),
+		fmt.Sprintf(`
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  tags = {
+    %[1]q = %[2]q
+  }
+}
+`, tag1Key, tag1Value))
+}
+
+func testAccTaskSetConfigTags2(rName, tag1Key, tag1Value, tag2Key, tag2Value string) string {
+	return acctest.ConfigCompose(
+		testAccTaskSetBaseConfig(rName),
+		fmt.Sprintf(`
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  tags = {
+    %[1]q = %[2]q
+    %[3]q = %[4]q
+  }
+}
+`, tag1Key, tag1Value, tag2Key, tag2Value))
+}
+
+func testAccTaskSet_withServiceRegistries(rName string) string {
+	return acctest.ConfigCompose(
+		acctest.ConfigAvailableAZsNoOptIn(),
+		fmt.Sprintf(`
+resource "aws_vpc" "test" {
+  cidr_block = "10.0.0.0/16"
+  tags = {
+    Name = "tf-acc-with-svc-reg"
+  }
+}
+
+resource "aws_subnet" "test" {
+  count             = 2
+  cidr_block        = cidrsubnet(aws_vpc.test.cidr_block, 8, count.index)
+  availability_zone = data.aws_availability_zones.available.names[count.index]
+  vpc_id            = aws_vpc.test.id
+  tags = {
+    Name = "tf-acc-with-svc-reg"
+  }
+}
+
+resource "aws_security_group" "test" {
+  name   = %[1]q
+  vpc_id = aws_vpc.test.id
+  ingress {
+    protocol    = "-1"
+    from_port   = 0
+    to_port     = 0
+    cidr_blocks = [aws_vpc.test.cidr_block]
+  }
+}
+
+resource "aws_service_discovery_private_dns_namespace" "test" {
+  name        = "%[1]s.terraform.local"
+  description = "test"
+  vpc         = aws_vpc.test.id
+}
+
+resource "aws_service_discovery_service" "test" {
+  name = %[1]q
+  dns_config {
+    namespace_id = aws_service_discovery_private_dns_namespace.test.id
+    dns_records {
+      ttl  = 5
+      type = "SRV"
+    }
+  }
+}
+
+resource "aws_ecs_cluster" "test" {
+  name = %[1]q
+}
+
+resource "aws_ecs_task_definition" "test" {
+  family                = %[1]q
+  network_mode          = "awsvpc"
+  container_definitions = <<DEFINITION
+[
+  {
+    "cpu": 128,
+    "essential": true,
+    "image": "mongo:latest",
+    "memory": 128,
+    "name": "mongodb"
+  }
+]
+DEFINITION
+}
+
+resource "aws_ecs_service" "test" {
+  name          = %[1]q
+  cluster       = aws_ecs_cluster.test.id
+  desired_count = 1
+  deployment_controller {
+    type = "EXTERNAL"
+  }
+}
+
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  service_registries {
+    port         = 34567
+    registry_arn = aws_service_discovery_service.test.arn
+  }
+  network_configuration {
+    security_groups = [aws_security_group.test.id]
+    subnets         = aws_subnet.test.*.id
+  }
+}
+`, rName))
+}
+
+func testAccTaskSetWithLaunchTypeFargate(rName string) string {
+	return acctest.ConfigCompose(
+		acctest.ConfigAvailableAZsNoOptIn(),
+		fmt.Sprintf(`
+resource "aws_vpc" "test" {
+  cidr_block = "10.10.0.0/16"
+  tags = {
+    Name = "terraform-testacc-ecs-service-with-launch-type-fargate"
+  }
+}
+
+resource "aws_subnet" "test" {
+  count             = 2
+  cidr_block        = cidrsubnet(aws_vpc.test.cidr_block, 8, count.index)
+  availability_zone = data.aws_availability_zones.available.names[count.index]
+  vpc_id            = aws_vpc.test.id
+  tags = {
+    Name = "tf-acc-ecs-service-with-launch-type-fargate"
+  }
+}
+
+resource "aws_security_group" "test" {
+  count = 2
+
+  name        = "%[1]s-${count.index}"
+  description = "Allow all inbound traffic"
+  vpc_id      = aws_vpc.test.id
+  ingress {
+    protocol    = "6"
+    from_port   = 80
+    to_port     = 8000
+    cidr_blocks = [aws_vpc.test.cidr_block]
+  }
+}
+
+resource "aws_ecs_cluster" "test" {
+  name = %[1]q
+}
+
+resource "aws_ecs_task_definition" "test" {
+  family                   = %[1]q
+  network_mode             = "awsvpc"
+  requires_compatibilities = ["FARGATE"]
+  cpu                      = "256"
+  memory                   = "512"
+  container_definitions    = <<DEFINITION
+[
+  {
+    "cpu": 256,
+    "essential": true,
+    "image": "mongo:latest",
+    "memory": 512,
+    "name": "mongodb",
+    "networkMode": "awsvpc"
+  }
+]
+DEFINITION
+}
+
+resource "aws_ecs_service" "test" {
+  name          = %[1]q
+  cluster       = aws_ecs_cluster.test.id
+  desired_count = 1
+  deployment_controller {
+    type = "EXTERNAL"
+  }
+}
+
+resource "aws_ecs_task_set" "test" {
+  service         = aws_ecs_service.test.id
+  cluster         = aws_ecs_cluster.test.id
+  task_definition = aws_ecs_task_definition.test.arn
+  launch_type     = "FARGATE"
+  network_configuration {
+    security_groups  = aws_security_group.test.*.id
+    subnets          = aws_subnet.test.*.id
+    assign_public_ip = false
+  }
+}
+`, rName))
+}
+
+func testAccTaskSetWithLaunchTypeFargateAndPlatformVersion(rName, platformVersion string) string {
+	return acctest.ConfigCompose(
+		acctest.ConfigAvailableAZsNoOptIn(),
+		fmt.Sprintf(`
+resource "aws_vpc" "test" {
+  cidr_block = "10.10.0.0/16"
+  tags = {
+    Name = "terraform-testacc-ecs-service-with-launch-type-fargate-and-platform-version"
+  }
+}
+
+resource "aws_subnet" "test" {
+  count             = 2
+  cidr_block        = cidrsubnet(aws_vpc.test.cidr_block, 8, count.index)
+  availability_zone = data.aws_availability_zones.available.names[count.index]
+  vpc_id            = aws_vpc.test.id
+  tags = {
+    Name = "tf-acc-ecs-service-with-launch-type-fargate-and-platform-version"
+  }
+}
+
+resource "aws_security_group" "test" {
+  count = 2
+
+  name        = "%[1]s-${count.index}"
+  description = "Allow all inbound traffic"
+  vpc_id      = aws_vpc.test.id
+  ingress {
+    protocol    = "6"
+    from_port   = 80
+    to_port     = 8000
+    cidr_blocks = [aws_vpc.test.cidr_block]
+  }
+}
+
+resource "aws_ecs_cluster" "test" {
+  name = %[1]q
+}
+
+resource "aws_ecs_task_definition" "test" {
+  family                   = %[1]q
+  network_mode             = "awsvpc"
+  requires_compatibilities = ["FARGATE"]
+  cpu                      = "256"
+  memory                   = "512"
+  container_definitions    = <<DEFINITION
+[
+  {
+    "cpu": 256,
+    "essential": true,
+    "image": "mongo:latest",
+    "memory": 512,
+    "name": "mongodb",
+    "networkMode": "awsvpc"
+  }
+]
+DEFINITION
+}
+
+resource "aws_ecs_service" "test" {
+  name          = %[1]q
+  cluster       = aws_ecs_cluster.test.id
+  desired_count = 1
+  deployment_controller {
+    type = "EXTERNAL"
+  }
+}
+
+resource "aws_ecs_task_set" "test" {
+  service          = aws_ecs_service.test.id
+  cluster          = aws_ecs_cluster.test.id
+  task_definition  = aws_ecs_task_definition.test.arn
+  launch_type      = "FARGATE"
+  platform_version = %[2]q
+  network_configuration {
+    security_groups  = aws_security_group.test.*.id
+    subnets          = aws_subnet.test.*.id
+    assign_public_ip = false
+  }
+}
+`, rName, platformVersion))
+}
+
+func testAccCheckTaskSetExists(name string) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		rs, ok := s.RootModule().Resources[name]
+		if !ok {
+			return fmt.Errorf("Not found: %s", name)
+		}
+
+		conn := acctest.Provider.Meta().(*conns.AWSClient).ECSConn
+
+		taskSetId, service, cluster, err := tfecs.TaskSetParseID(rs.Primary.ID)
+
+		if err != nil {
+			return err
+		}
+
+		input := &ecs.DescribeTaskSetsInput{
+			TaskSets: aws.StringSlice([]string{taskSetId}),
+			Cluster:  aws.String(cluster),
+			Service:  aws.String(service),
+		}
+
+		output, err := conn.DescribeTaskSets(input)
+
+		if err != nil {
+			return err
+		}
+
+		if output == nil || len(output.TaskSets) == 0 {
+			return fmt.Errorf("ECS TaskSet (%s) not found", rs.Primary.ID)
+		}
+
+		return nil
+	}
+}
+
+func testAccCheckTaskSetDestroy(s *terraform.State) error {
+	conn := acctest.Provider.Meta().(*conns.AWSClient).ECSConn
+
+	for _, rs := range s.RootModule().Resources {
+		if rs.Type != "aws_ecs_task_set" {
+			continue
+		}
+
+		taskSetId, service, cluster, err := tfecs.TaskSetParseID(rs.Primary.ID)
+
+		if err != nil {
+			return err
+		}
+
+		input := &ecs.DescribeTaskSetsInput{
+			TaskSets: aws.StringSlice([]string{taskSetId}),
+			Cluster:  aws.String(cluster),
+			Service:  aws.String(service),
+		}
+
+		output, err := conn.DescribeTaskSets(input)
+
+		if tfawserr.ErrCodeEquals(err, ecs.ErrCodeClusterNotFoundException, ecs.ErrCodeServiceNotFoundException, ecs.ErrCodeTaskSetNotFoundException) {
+			continue
+		}
+
+		if err != nil {
+			return err
+		}
+
+		if output != nil && len(output.TaskSets) == 1 {
+			return fmt.Errorf("ECS TaskSet (%s) still exists", rs.Primary.ID)
+		}
+	}
+
+	return nil
+}
diff --git a/internal/service/ecs/wait.go b/internal/service/ecs/wait.go
index 85365488105..1ede62d9374 100644
--- a/internal/service/ecs/wait.go
+++ b/internal/service/ecs/wait.go
@@ -21,6 +21,9 @@ const (
 	clusterAvailableTimeout = 10 * time.Minute
 	clusterDeleteTimeout    = 10 * time.Minute
 	clusterAvailableDelay   = 10 * time.Second
+
+	taskSetCreateTimeout = 10 * time.Minute
+	taskSetDeleteTimeout = 10 * time.Minute
 )
 
 func waitCapacityProviderDeleted(conn *ecs.ECS, arn string) (*ecs.CapacityProvider, error) {
@@ -153,3 +156,29 @@ func waitClusterDeleted(conn *ecs.ECS, arn string) (*ecs.Cluster, error) {
 
 	return nil, err
 }
+
+func waitTaskSetStable(conn *ecs.ECS, timeout time.Duration, taskSetID, service, cluster string) error {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{ecs.StabilityStatusStabilizing},
+		Target:  []string{ecs.StabilityStatusSteadyState},
+		Refresh: stabilityStatusTaskSet(conn, taskSetID, service, cluster),
+		Timeout: timeout,
+	}
+
+	_, err := stateConf.WaitForState()
+
+	return err
+}
+
+func waitTaskSetDeleted(conn *ecs.ECS, taskSetID, service, cluster string) error {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{taskSetStatusActive, taskSetStatusPrimary, taskSetStatusDraining},
+		Target:  []string{},
+		Refresh: statusTaskSet(conn, taskSetID, service, cluster),
+		Timeout: taskSetDeleteTimeout,
+	}
+
+	_, err := stateConf.WaitForState()
+
+	return err
+}
diff --git a/website/docs/r/ecs_task_set.html.markdown b/website/docs/r/ecs_task_set.html.markdown
new file mode 100644
index 00000000000..13cd179e115
--- /dev/null
+++ b/website/docs/r/ecs_task_set.html.markdown
@@ -0,0 +1,136 @@
+---
+subcategory: "ECS"
+layout: "aws"
+page_title: "AWS: aws_ecs_task_set"
+description: |-
+  Provides an ECS task set.
+---
+
+# Resource: aws_ecs_task_set
+
+Provides an ECS task set - effectively a task that is expected to run until an error occurs or a user terminates it (typically a webserver or a database).
+
+See [ECS Task Set section in AWS developer guide](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/deployment-type-external.html).
+
+## Example Usage
+
+```terraform
+resource "aws_ecs_task_set" "example" {
+  service         = aws_ecs_service.example.id
+  cluster         = aws_ecs_cluster.example.id
+  task_definition = aws_ecs_task_definition.example.arn
+
+  load_balancer {
+    target_group_arn = aws_lb_target_group.example.arn
+    container_name   = "mongo"
+    container_port   = 8080
+  }
+}
+```
+
+### Ignoring Changes to Scale
+
+You can utilize the generic Terraform resource [lifecycle configuration block](https://www.terraform.io/docs/configuration/meta-arguments/lifecycle.html) with `ignore_changes` to create an ECS service with an initial count of running instances, then ignore any changes to that count caused externally (e.g. Application Autoscaling).
+
+```terraform
+resource "aws_ecs_task_set" "example" {
+  # ... other configurations ...
+
+  # Example: Run 50% of the servcie's desired count
+  scale {
+    value = 50.0
+  }
+
+  # Optional: Allow external changes without Terraform plan difference
+  lifecycle {
+    ignore_changes = ["scale"]
+  }
+}
+```
+
+## Argument Reference
+
+The following arguments are required:
+
+* `service` - (Required) The short name or ARN of the ECS service.
+* `cluster` - (Required) The short name or ARN of the cluster that hosts the service to create the task set in.
+* `task_definition` - (Required) The family and revision (`family:revision`) or full ARN of the task definition that you want to run in your service.
+
+The following arguments are optional:
+
+* `capacity_provider_strategy` - (Optional) The capacity provider strategy to use for the service. Can be one or more.  [Defined below](#capacity_provider_strategy).
+* `external_id` - (Optional) The external ID associated with the task set.
+* `force_delete` - (Optional) Whether to allow deleting the task set without waiting for scaling down to 0. You can force a task set to delete even if it's in the process of scaling a resource. Normally, Terraform drains all the tasks before deleting the task set. This bypasses that behavior and potentially leaves resources dangling.
+* `launch_type` - (Optional) The launch type on which to run your service. The valid values are `EC2`, `FARGATE`, and `EXTERNAL`. Defaults to `EC2`.
+* `load_balancer` - (Optional) Details on load balancers that are used with a task set. [Detailed below](#load_balancer).
+* `platform_version` - (Optional) The platform version on which to run your service. Only applicable for `launch_type` set to `FARGATE`. Defaults to `LATEST`. More information about Fargate platform versions can be found in the [AWS ECS User Guide](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/platform_versions.html).
+* `network_configuration` - (Optional) The network configuration for the service. This parameter is required for task definitions that use the `awsvpc` network mode to receive their own Elastic Network Interface, and it is not supported for other network modes. [Detailed below](#network_configuration).
+* `scale` - (Optional) A floating-point percentage of the desired number of tasks to place and keep running in the task set. [Detailed below](#scale).
+* `service_registries` - (Optional) The service discovery registries for the service. The maximum number of `service_registries` blocks is `1`. [Detailed below](#service_registries).
+* `tags` - (Optional) A map of tags to assign to the file system. If configured with a provider [`default_tags` configuration block](/docs/providers/aws/index.html#default_tags-configuration-block) present, tags with matching keys will overwrite those defined at the provider-level. If you have set `copy_tags_to_backups` to true, and you specify one or more tags, no existing file system tags are copied from the file system to the backup.
+* `wait_until_stable` - (Optional) Whether `terraform` should wait until the task set has reached `STEADY_STATE`.
+* `wait_until_stable_timeout` - (Optional) Wait timeout for task set to reach `STEADY_STATE`. Valid time units include `ns`, `us` (or `µs`), `ms`, `s`, `m`, and `h`. Default `10m`.
+
+## capacity_provider_strategy
+
+The `capacity_provider_strategy` configuration block supports the following:
+
+* `capacity_provider` - (Required) The short name or full Amazon Resource Name (ARN) of the capacity provider.
+* `weight` - (Required) The relative percentage of the total number of launched tasks that should use the specified capacity provider.
+* `base` - (Optional) The number of tasks, at a minimum, to run on the specified capacity provider. Only one capacity provider in a capacity provider strategy can have a base defined.
+
+## load_balancer
+
+The `load_balancer` configuration block supports the following:
+
+* `container_name` - (Required) The name of the container to associate with the load balancer (as it appears in a container definition).
+* `load_balancer_name` - (Optional, Required for ELB Classic) The name of the ELB (Classic) to associate with the service.
+* `target_group_arn` - (Optional, Required for ALB/NLB) The ARN of the Load Balancer target group to associate with the service.
+* `container_port` - (Optional) The port on the container to associate with the load balancer. Defaults to `0` if not specified.
+
+~> **Note:** Specifying multiple `load_balancer` configurations is still not supported by AWS for ECS task set.
+
+## network_configuration
+
+The `network_configuration` configuration block supports the following:
+
+* `subnets` - (Required) The subnets associated with the task or service. Maximum of 16.
+* `security_groups` - (Optional) The security groups associated with the task or service. If you do not specify a security group, the default security group for the VPC is used. Maximum of 5.
+* `assign_public_ip` - (Optional) Whether to assign a public IP address to the ENI (`FARGATE` launch type only). Valid values are `true` or `false`. Default `false`.
+
+For more information, see [Task Networking](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-networking.html).
+
+## scale
+
+The `scale` configuration block supports the following:
+
+* `unit` - (Optional) The unit of measure for the scale value. Default: `PERCENT`.
+* `value` - (Optional) The value, specified as a percent total of a service's `desiredCount`, to scale the task set. Defaults to `0` if not specified. Accepted values are numbers between 0.0 and 100.0.
+
+## service_registries
+
+`service_registries` support the following:
+
+* `registry_arn` - (Required) The ARN of the Service Registry. The currently supported service registry is Amazon Route 53 Auto Naming Service([`aws_service_discovery_service` resource](/docs/providers/aws/r/service_discovery_service.html)). For more information, see [Service](https://docs.aws.amazon.com/Route53/latest/APIReference/API_autonaming_Service.html).
+* `port` - (Optional) The port value used if your Service Discovery service specified an SRV record.
+* `container_port` - (Optional) The port value, already specified in the task definition, to be used for your service discovery service.
+* `container_name` - (Optional) The container name value, already specified in the task definition, to be used for your service discovery service.
+
+## Attributes Reference
+
+In addition to all arguments above, the following attributes are exported:
+
+* `id` - The `task_set_id`, `service` and `cluster` separated by commas (`,`).
+* `arn` - The Amazon Resource Name (ARN) that identifies the task set.
+* `stability_status` - The stability status. This indicates whether the task set has reached a steady state.
+* `status` - The status of the task set.
+* `tags_all` - A map of tags assigned to the resource, including those inherited from the provider [`default_tags` configuration block](/docs/providers/aws/index.html#default_tags-configuration-block).
+* `task_set_id` - The ID of the task set.
+
+## Import
+
+ECS Task Sets can be imported via the `task_set_id`, `service`, and `cluster` separated by commas (`,`) e.g.
+
+```
+$ terraform import aws_ecs_task_set.example ecs-svc/7177320696926227436,arn:aws:ecs:us-west-2:123456789101:service/example/example-1234567890,arn:aws:ecs:us-west-2:123456789101:cluster/example
+```