diff --git a/.changelog/27574.txt b/.changelog/27574.txt
new file mode 100644
index 000000000000..a6339f407380
--- /dev/null
+++ b/.changelog/27574.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_sfn_state_machine: Add `name_prefix` argument
+```
\ No newline at end of file
diff --git a/internal/acctest/acctest.go b/internal/acctest/acctest.go
index 25008e4683b3..0f2f4d52ae45 100644
--- a/internal/acctest/acctest.go
+++ b/internal/acctest/acctest.go
@@ -1780,18 +1780,9 @@ func ConfigLatestAmazonLinux2HVMEBSARM64AMI() string {
 }
 
 func ConfigLambdaBase(policyName, roleName, sgName string) string {
-	return fmt.Sprintf(`
+	return ConfigCompose(ConfigAvailableAZsNoOptIn(), fmt.Sprintf(`
 data "aws_partition" "current" {}
 
-data "aws_availability_zones" "available" {
-  state = "available"
-
-  filter {
-    name   = "opt-in-status"
-    values = ["opt-in-not-required"]
-  }
-}
-
 resource "aws_iam_role_policy" "iam_policy_for_lambda" {
   name = "%s"
   role = aws_iam_role.iam_for_lambda.id
@@ -1912,7 +1903,7 @@ resource "aws_security_group" "sg_for_lambda" {
     cidr_blocks = ["0.0.0.0/0"]
   }
 }
-`, policyName, roleName, sgName)
+`, policyName, roleName, sgName))
 }
 
 func ConfigVPCWithSubnets(rName string, subnetCount int) string {
diff --git a/internal/service/sfn/activity.go b/internal/service/sfn/activity.go
index 0e4b35f93edb..1d20ded60740 100644
--- a/internal/service/sfn/activity.go
+++ b/internal/service/sfn/activity.go
@@ -1,42 +1,45 @@
 package sfn
 
 import (
-	"fmt"
+	"context"
 	"log"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/service/sfn"
+	"github.com/hashicorp/aws-sdk-go-base/v2/awsv1shim/v2/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/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"
 	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 ResourceActivity() *schema.Resource {
 	return &schema.Resource{
-		Create: resourceActivityCreate,
-		Read:   resourceActivityRead,
-		Update: resourceActivityUpdate,
-		Delete: resourceActivityDelete,
+		CreateWithoutTimeout: resourceActivityCreate,
+		ReadWithoutTimeout:   resourceActivityRead,
+		UpdateWithoutTimeout: resourceActivityUpdate,
+		DeleteWithoutTimeout: resourceActivityDelete,
+
 		Importer: &schema.ResourceImporter{
 			State: schema.ImportStatePassthrough,
 		},
 
 		Schema: map[string]*schema.Schema{
+			"creation_date": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
 			"name": {
 				Type:         schema.TypeString,
 				Required:     true,
 				ForceNew:     true,
 				ValidateFunc: validation.StringLenBetween(0, 80),
 			},
-
-			"creation_date": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
 			"tags":     tftags.TagsSchema(),
 			"tags_all": tftags.TagsSchemaComputed(),
 		},
@@ -45,97 +48,118 @@ func ResourceActivity() *schema.Resource {
 	}
 }
 
-func resourceActivityCreate(d *schema.ResourceData, meta interface{}) error {
+func resourceActivityCreate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
 	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
 	tags := defaultTagsConfig.MergeTags(tftags.New(d.Get("tags").(map[string]interface{})))
-	log.Print("[DEBUG] Creating Step Function Activity")
 
-	params := &sfn.CreateActivityInput{
-		Name: aws.String(d.Get("name").(string)),
+	name := d.Get("name").(string)
+	input := &sfn.CreateActivityInput{
+		Name: aws.String(name),
 		Tags: Tags(tags.IgnoreAWS()),
 	}
 
-	activity, err := conn.CreateActivity(params)
+	output, err := conn.CreateActivityWithContext(ctx, input)
+
 	if err != nil {
-		return fmt.Errorf("Error creating Step Function Activity: %s", err)
+		return diag.Errorf("creating Step Functions Activity (%s): %s", name, err)
 	}
 
-	d.SetId(aws.StringValue(activity.ActivityArn))
+	d.SetId(aws.StringValue(output.ActivityArn))
 
-	return resourceActivityRead(d, meta)
+	return resourceActivityRead(ctx, d, meta)
 }
 
-func resourceActivityUpdate(d *schema.ResourceData, meta interface{}) error {
-	conn := meta.(*conns.AWSClient).SFNConn
-
-	if d.HasChange("tags_all") {
-		o, n := d.GetChange("tags_all")
-		if err := UpdateTags(conn, d.Id(), o, n); err != nil {
-			return fmt.Errorf("error updating tags: %s", err)
-		}
-	}
-
-	return resourceActivityRead(d, meta)
-}
-
-func resourceActivityRead(d *schema.ResourceData, meta interface{}) error {
+func resourceActivityRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
 	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
 	ignoreTagsConfig := meta.(*conns.AWSClient).IgnoreTagsConfig
 
-	log.Printf("[DEBUG] Reading Step Function Activity: %s", d.Id())
+	output, err := FindActivityByARN(ctx, conn, d.Id())
 
-	sm, err := conn.DescribeActivity(&sfn.DescribeActivityInput{
-		ActivityArn: aws.String(d.Id()),
-	})
-	if err != nil {
-		if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == "ActivityDoesNotExist" {
-			d.SetId("")
-			return nil
-		}
-		return err
+	if !d.IsNewResource() && tfresource.NotFound(err) {
+		log.Printf("[WARN] Step Functions Activity (%s) not found, removing from state", d.Id())
+		d.SetId("")
+		return nil
 	}
 
-	d.Set("name", sm.Name)
-
-	if err := d.Set("creation_date", sm.CreationDate.Format(time.RFC3339)); err != nil {
-		log.Printf("[DEBUG] Error setting creation_date: %s", err)
+	if err != nil {
+		return diag.Errorf("reading Step Functions Activity (%s): %s", d.Id(), err)
 	}
 
-	tags, err := ListTags(conn, d.Id())
+	d.Set("creation_date", output.CreationDate.Format(time.RFC3339))
+	d.Set("name", output.Name)
+
+	tags, err := ListTagsWithContext(ctx, conn, d.Id())
 
 	if err != nil {
-		return fmt.Errorf("error listing tags for SFN Activity (%s): %s", d.Id(), err)
+		return diag.Errorf("listing tags for Step Functions Activity (%s): %s", d.Id(), err)
 	}
 
 	tags = 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)
+		return diag.Errorf("setting tags: %s", err)
 	}
 
 	if err := d.Set("tags_all", tags.Map()); err != nil {
-		return fmt.Errorf("error setting tags_all: %w", err)
+		return diag.Errorf("setting tags_all: %s", err)
 	}
 
 	return nil
 }
 
-func resourceActivityDelete(d *schema.ResourceData, meta interface{}) error {
+func resourceActivityUpdate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
-	log.Printf("[DEBUG] Deleting Step Functions Activity: %s", d.Id())
 
-	input := &sfn.DeleteActivityInput{
-		ActivityArn: aws.String(d.Id()),
+	if d.HasChange("tags_all") {
+		o, n := d.GetChange("tags_all")
+
+		if err := UpdateTagsWithContext(ctx, conn, d.Id(), o, n); err != nil {
+			return diag.Errorf("updating Step Functions Activity (%s) tags: %s", d.Id(), err)
+		}
 	}
 
-	_, err := conn.DeleteActivity(input)
+	return resourceActivityRead(ctx, d, meta)
+}
+
+func resourceActivityDelete(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	conn := meta.(*conns.AWSClient).SFNConn
+
+	log.Printf("[DEBUG] Deleting Step Functions Activity: %s", d.Id())
+	_, err := conn.DeleteActivityWithContext(ctx, &sfn.DeleteActivityInput{
+		ActivityArn: aws.String(d.Id()),
+	})
 
 	if err != nil {
-		return fmt.Errorf("Error deleting SFN Activity: %s", err)
+		return diag.Errorf("deleting Step Functions Activity (%s): %s", d.Id(), err)
 	}
 
 	return nil
 }
+
+func FindActivityByARN(ctx context.Context, conn *sfn.SFN, arn string) (*sfn.DescribeActivityOutput, error) {
+	input := &sfn.DescribeActivityInput{
+		ActivityArn: aws.String(arn),
+	}
+
+	output, err := conn.DescribeActivityWithContext(ctx, input)
+
+	if tfawserr.ErrCodeEquals(err, sfn.ErrCodeActivityDoesNotExist) {
+		return nil, &resource.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.CreationDate == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output, nil
+}
diff --git a/internal/service/sfn/activity_data_source.go b/internal/service/sfn/activity_data_source.go
index 69288fe1f9c7..f6d043825e20 100644
--- a/internal/service/sfn/activity_data_source.go
+++ b/internal/service/sfn/activity_data_source.go
@@ -1,22 +1,22 @@
 package sfn
 
 import (
-	"fmt"
-	"log"
+	"context"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sfn"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
 )
 
 func DataSourceActivity() *schema.Resource {
 	return &schema.Resource{
-		Read: dataSourceActivityRead,
+		ReadWithoutTimeout: dataSourceActivityRead,
 
 		Schema: map[string]*schema.Schema{
-			"name": {
+			"arn": {
 				Type:     schema.TypeString,
 				Computed: true,
 				Optional: true,
@@ -25,7 +25,11 @@ func DataSourceActivity() *schema.Resource {
 					"name",
 				},
 			},
-			"arn": {
+			"creation_date": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"name": {
 				Type:     schema.TypeString,
 				Computed: true,
 				Optional: true,
@@ -34,75 +38,61 @@ func DataSourceActivity() *schema.Resource {
 					"name",
 				},
 			},
-			"creation_date": {
-				Type:     schema.TypeString,
-				Computed: true,
-			},
 		},
 	}
 }
 
-func dataSourceActivityRead(d *schema.ResourceData, meta interface{}) error {
-	client := meta.(*conns.AWSClient)
-	conn := client.SFNConn
-	log.Print("[DEBUG] Reading Step Function Activity")
+func dataSourceActivityRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	conn := meta.(*conns.AWSClient).SFNConn
 
-	if nm, ok := d.GetOk("name"); ok {
-		name := nm.(string)
-		var acts []*sfn.ActivityListItem
+	if v, ok := d.GetOk("name"); ok {
+		name := v.(string)
+		var activities []*sfn.ActivityListItem
 
-		err := conn.ListActivitiesPages(&sfn.ListActivitiesInput{}, func(page *sfn.ListActivitiesOutput, lastPage bool) bool {
-			for _, a := range page.Activities {
-				if name == aws.StringValue(a.Name) {
-					acts = append(acts, a)
+		err := conn.ListActivitiesPagesWithContext(ctx, &sfn.ListActivitiesInput{}, func(page *sfn.ListActivitiesOutput, lastPage bool) bool {
+			if page == nil {
+				return !lastPage
+			}
+
+			for _, v := range page.Activities {
+				if name == aws.StringValue(v.Name) {
+					activities = append(activities, v)
 				}
 			}
+
 			return !lastPage
 		})
 
 		if err != nil {
-			return fmt.Errorf("Error listing activities: %w", err)
+			return diag.Errorf("listing Step Functions Activities: %s", err)
 		}
 
-		if len(acts) == 0 {
-			return fmt.Errorf("No activity found with name %s in this region", name)
+		if n := len(activities); n == 0 {
+			return diag.Errorf("no Step Functions Activities matched")
+		} else if n > 1 {
+			return diag.Errorf("%d Step Functions Activities matched; use additional constraints to reduce matches to a single Activity", n)
 		}
 
-		if len(acts) > 1 {
-			return fmt.Errorf("Found more than 1 activity with name %s in this region", name)
-		}
-
-		act := acts[0]
-
-		d.SetId(aws.StringValue(act.ActivityArn))
-		d.Set("name", act.Name)
-		d.Set("arn", act.ActivityArn)
-		if err := d.Set("creation_date", act.CreationDate.Format(time.RFC3339)); err != nil {
-			log.Printf("[DEBUG] Error setting creation_date: %s", err)
-		}
-	}
+		activity := activities[0]
 
-	if rnm, ok := d.GetOk("arn"); ok {
-		arn := rnm.(string)
-		params := &sfn.DescribeActivityInput{
-			ActivityArn: aws.String(arn),
-		}
+		arn := aws.StringValue(activity.ActivityArn)
+		d.SetId(arn)
+		d.Set("arn", arn)
+		d.Set("creation_date", activity.CreationDate.Format(time.RFC3339))
+		d.Set("name", activity.Name)
+	} else if v, ok := d.GetOk("arn"); ok {
+		arn := v.(string)
+		activity, err := FindActivityByARN(ctx, conn, arn)
 
-		act, err := conn.DescribeActivity(params)
 		if err != nil {
-			return fmt.Errorf("Error describing activities: %w", err)
+			return diag.Errorf("reading Step Functions Activity (%s): %s", arn, err)
 		}
 
-		if act == nil {
-			return fmt.Errorf("No activity found with arn %s in this region", arn)
-		}
-
-		d.SetId(aws.StringValue(act.ActivityArn))
-		d.Set("name", act.Name)
-		d.Set("arn", act.ActivityArn)
-		if err := d.Set("creation_date", act.CreationDate.Format(time.RFC3339)); err != nil {
-			log.Printf("[DEBUG] Error setting creation_date: %s", err)
-		}
+		arn = aws.StringValue(activity.ActivityArn)
+		d.SetId(arn)
+		d.Set("arn", arn)
+		d.Set("creation_date", activity.CreationDate.Format(time.RFC3339))
+		d.Set("name", activity.Name)
 	}
 
 	return nil
diff --git a/internal/service/sfn/activity_data_source_test.go b/internal/service/sfn/activity_data_source_test.go
index 1abab8e100db..ef44c2f29462 100644
--- a/internal/service/sfn/activity_data_source_test.go
+++ b/internal/service/sfn/activity_data_source_test.go
@@ -10,10 +10,11 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/acctest"
 )
 
-func TestAccSFNActivityDataSource_StepFunctions_basic(t *testing.T) {
+func TestAccSFNActivityDataSource_basic(t *testing.T) {
 	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
 	resourceName := "aws_sfn_activity.test"
-	dataName := "data.aws_sfn_activity.test"
+	dataSource1Name := "data.aws_sfn_activity.by_name"
+	dataSource2Name := "data.aws_sfn_activity.by_arn"
 
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:                 func() { acctest.PreCheck(t) },
@@ -21,45 +22,33 @@ func TestAccSFNActivityDataSource_StepFunctions_basic(t *testing.T) {
 		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccActivityDataSourceConfig_checkARN(rName),
-				Check: resource.ComposeTestCheckFunc(
-					resource.TestCheckResourceAttrPair(resourceName, "id", dataName, "id"),
-					resource.TestCheckResourceAttrPair(resourceName, "creation_date", dataName, "creation_date"),
-					resource.TestCheckResourceAttrPair(resourceName, "name", dataName, "name"),
-				),
-			},
-			{
-				Config: testAccActivityDataSourceConfig_checkName(rName),
-				Check: resource.ComposeTestCheckFunc(
-					resource.TestCheckResourceAttrPair(resourceName, "id", dataName, "id"),
-					resource.TestCheckResourceAttrPair(resourceName, "creation_date", dataName, "creation_date"),
-					resource.TestCheckResourceAttrPair(resourceName, "name", dataName, "name"),
+				Config: testAccActivityDataSourceConfig_basic(rName),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttrPair(resourceName, "id", dataSource1Name, "arn"),
+					resource.TestCheckResourceAttrPair(resourceName, "creation_date", dataSource1Name, "creation_date"),
+					resource.TestCheckResourceAttrPair(resourceName, "name", dataSource1Name, "name"),
+
+					resource.TestCheckResourceAttrPair(resourceName, "id", dataSource2Name, "arn"),
+					resource.TestCheckResourceAttrPair(resourceName, "creation_date", dataSource2Name, "creation_date"),
+					resource.TestCheckResourceAttrPair(resourceName, "name", dataSource2Name, "name"),
 				),
 			},
 		},
 	})
 }
 
-func testAccActivityDataSourceConfig_checkARN(rName string) string {
+func testAccActivityDataSourceConfig_basic(rName string) string {
 	return fmt.Sprintf(`
 resource aws_sfn_activity "test" {
-  name = "%s"
+  name = %[1]q
 }
 
-data aws_sfn_activity "test" {
-  arn = aws_sfn_activity.test.id
-}
-`, rName)
-}
-
-func testAccActivityDataSourceConfig_checkName(rName string) string {
-	return fmt.Sprintf(`
-resource aws_sfn_activity "test" {
-  name = "%s"
+data aws_sfn_activity "by_name" {
+  name = aws_sfn_activity.test.name
 }
 
-data aws_sfn_activity "test" {
-  name = aws_sfn_activity.test.name
+data aws_sfn_activity "by_arn" {
+  arn = aws_sfn_activity.test.id
 }
 `, rName)
 }
diff --git a/internal/service/sfn/activity_test.go b/internal/service/sfn/activity_test.go
index f3a51d6d78a8..c3b15c3c60b4 100644
--- a/internal/service/sfn/activity_test.go
+++ b/internal/service/sfn/activity_test.go
@@ -1,22 +1,23 @@
 package sfn_test
 
 import (
+	"context"
 	"fmt"
 	"testing"
 	"time"
 
-	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sfn"
-	"github.com/hashicorp/aws-sdk-go-base/v2/awsv1shim/v2/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"
+	tfsfn "github.com/hashicorp/terraform-provider-aws/internal/service/sfn"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
 func TestAccSFNActivity_basic(t *testing.T) {
-	name := sdkacctest.RandString(10)
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
 	resourceName := "aws_sfn_activity.test"
 
 	resource.ParallelTest(t, resource.TestCase{
@@ -26,11 +27,12 @@ func TestAccSFNActivity_basic(t *testing.T) {
 		CheckDestroy:             testAccCheckActivityDestroy,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccActivityConfig_basic(name),
+				Config: testAccActivityConfig_basic(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckActivityExists(resourceName),
-					resource.TestCheckResourceAttr(resourceName, "name", name),
 					resource.TestCheckResourceAttrSet(resourceName, "creation_date"),
+					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "0"),
 				),
 			},
 			{
@@ -42,8 +44,30 @@ func TestAccSFNActivity_basic(t *testing.T) {
 	})
 }
 
+func TestAccSFNActivity_disappears(t *testing.T) {
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_sfn_activity.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(t) },
+		ErrorCheck:               acctest.ErrorCheck(t, sfn.EndpointsID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckActivityDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccActivityConfig_basic(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckActivityExists(resourceName),
+					acctest.CheckResourceDisappears(acctest.Provider, tfsfn.ResourceActivity(), resourceName),
+				),
+				ExpectNonEmptyPlan: true,
+			},
+		},
+	})
+}
+
 func TestAccSFNActivity_tags(t *testing.T) {
-	name := sdkacctest.RandString(10)
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
 	resourceName := "aws_sfn_activity.test"
 
 	resource.ParallelTest(t, resource.TestCase{
@@ -53,7 +77,7 @@ func TestAccSFNActivity_tags(t *testing.T) {
 		CheckDestroy:             testAccCheckActivityDestroy,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccActivityConfig_basicTags1(name, "key1", "value1"),
+				Config: testAccActivityConfig_basicTags1(rName, "key1", "value1"),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckActivityExists(resourceName),
 					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
@@ -66,7 +90,7 @@ func TestAccSFNActivity_tags(t *testing.T) {
 				ImportStateVerify: true,
 			},
 			{
-				Config: testAccActivityConfig_basicTags2(name, "key1", "value1updated", "key2", "value2"),
+				Config: testAccActivityConfig_basicTags2(rName, "key1", "value1updated", "key2", "value2"),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckActivityExists(resourceName),
 					resource.TestCheckResourceAttr(resourceName, "tags.%", "2"),
@@ -75,7 +99,7 @@ func TestAccSFNActivity_tags(t *testing.T) {
 				),
 			},
 			{
-				Config: testAccActivityConfig_basicTags1(name, "key2", "value2"),
+				Config: testAccActivityConfig_basicTags1(rName, "key2", "value2"),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckActivityExists(resourceName),
 					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
@@ -94,14 +118,12 @@ func testAccCheckActivityExists(n string) resource.TestCheckFunc {
 		}
 
 		if rs.Primary.ID == "" {
-			return fmt.Errorf("No Step Function ID set")
+			return fmt.Errorf("No Step Functions Activity ID set")
 		}
 
 		conn := acctest.Provider.Meta().(*conns.AWSClient).SFNConn
 
-		_, err := conn.DescribeActivity(&sfn.DescribeActivityInput{
-			ActivityArn: aws.String(rs.Primary.ID),
-		})
+		_, err := tfsfn.FindActivityByARN(context.Background(), conn, rs.Primary.ID)
 
 		return err
 	}
@@ -115,15 +137,11 @@ func testAccCheckActivityDestroy(s *terraform.State) error {
 			continue
 		}
 
-		// Retrying as Read after Delete is not always consistent
-		retryErr := resource.Retry(1*time.Minute, func() *resource.RetryError {
-			var err error
+		// Retrying as Read after Delete is not always consistent.
+		err := resource.Retry(1*time.Minute, func() *resource.RetryError {
+			_, err := tfsfn.FindActivityByARN(context.Background(), conn, rs.Primary.ID)
 
-			_, err = conn.DescribeActivity(&sfn.DescribeActivityInput{
-				ActivityArn: aws.String(rs.Primary.ID),
-			})
-
-			if tfawserr.ErrCodeEquals(err, sfn.ErrCodeActivityDoesNotExist) {
+			if tfresource.NotFound(err) {
 				return nil
 			}
 
@@ -131,21 +149,19 @@ func testAccCheckActivityDestroy(s *terraform.State) error {
 				return resource.NonRetryableError(err)
 			}
 
-			// If there are no errors, the removal failed
-			// and the object is not yet removed.
-			return resource.RetryableError(fmt.Errorf("Expected AWS Step Function Activity to be destroyed, but was still found, retrying"))
+			return resource.RetryableError(fmt.Errorf("Step Functions Activity still exists: %s", rs.Primary.ID))
 		})
 
-		return retryErr
+		return err
 	}
 
-	return fmt.Errorf("Default error in Step Function Test")
+	return nil
 }
 
 func testAccActivityConfig_basic(rName string) string {
 	return fmt.Sprintf(`
 resource "aws_sfn_activity" "test" {
-  name = "%s"
+  name = %[1]q
 }
 `, rName)
 }
@@ -153,10 +169,10 @@ resource "aws_sfn_activity" "test" {
 func testAccActivityConfig_basicTags1(rName, tag1Key, tag1Value string) string {
 	return fmt.Sprintf(`
 resource "aws_sfn_activity" "test" {
-  name = "%s"
+  name = %[1]q
 
   tags = {
-    %q = %q
+    %[2]q = %[3]q
   }
 }
 `, rName, tag1Key, tag1Value)
@@ -165,11 +181,11 @@ resource "aws_sfn_activity" "test" {
 func testAccActivityConfig_basicTags2(rName, tag1Key, tag1Value, tag2Key, tag2Value string) string {
 	return fmt.Sprintf(`
 resource "aws_sfn_activity" "test" {
-  name = "%s"
+  name = %[1]q
 
   tags = {
-    %q = %q
-    %q = %q
+    %[2]q = %[3]q
+    %[4]q = %[5]q
   }
 }
 `, rName, tag1Key, tag1Value, tag2Key, tag2Value)
diff --git a/internal/service/sfn/find.go b/internal/service/sfn/find.go
deleted file mode 100644
index 071c78774f81..000000000000
--- a/internal/service/sfn/find.go
+++ /dev/null
@@ -1,36 +0,0 @@
-package sfn
-
-import (
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/sfn"
-	"github.com/hashicorp/aws-sdk-go-base/v2/awsv1shim/v2/tfawserr"
-	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
-)
-
-func FindStateMachineByARN(conn *sfn.SFN, arn string) (*sfn.DescribeStateMachineOutput, error) {
-	input := &sfn.DescribeStateMachineInput{
-		StateMachineArn: aws.String(arn),
-	}
-
-	output, err := conn.DescribeStateMachine(input)
-
-	if tfawserr.ErrCodeEquals(err, sfn.ErrCodeStateMachineDoesNotExist) {
-		return nil, &resource.NotFoundError{
-			LastError:   err,
-			LastRequest: input,
-		}
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	if output == nil {
-		return nil, &resource.NotFoundError{
-			Message:     "Empty result",
-			LastRequest: input,
-		}
-	}
-
-	return output, nil
-}
diff --git a/internal/service/sfn/state_machine.go b/internal/service/sfn/state_machine.go
index dae130ee0fa6..ea1a3f5f1328 100644
--- a/internal/service/sfn/state_machine.go
+++ b/internal/service/sfn/state_machine.go
@@ -1,17 +1,21 @@
 package sfn
 
 import (
+	"context"
 	"fmt"
 	"log"
+	"regexp"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sfn"
 	"github.com/hashicorp/aws-sdk-go-base/v2/awsv1shim/v2/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/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"
+	"github.com/hashicorp/terraform-provider-aws/internal/create"
 	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"
@@ -19,10 +23,11 @@ import (
 
 func ResourceStateMachine() *schema.Resource {
 	return &schema.Resource{
-		Create: resourceStateMachineCreate,
-		Read:   resourceStateMachineRead,
-		Update: resourceStateMachineUpdate,
-		Delete: resourceStateMachineDelete,
+		CreateWithoutTimeout: resourceStateMachineCreate,
+		ReadWithoutTimeout:   resourceStateMachineRead,
+		UpdateWithoutTimeout: resourceStateMachineUpdate,
+		DeleteWithoutTimeout: resourceStateMachineDelete,
+
 		Importer: &schema.ResourceImporter{
 			State: schema.ImportStatePassthrough,
 		},
@@ -32,18 +37,15 @@ func ResourceStateMachine() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-
 			"creation_date": {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-
 			"definition": {
 				Type:         schema.TypeString,
 				Required:     true,
 				ValidateFunc: validation.StringLenBetween(0, 1024*1024), // 1048576
 			},
-
 			"logging_configuration": {
 				Type:     schema.TypeList,
 				Optional: true,
@@ -68,36 +70,39 @@ func ResourceStateMachine() *schema.Resource {
 				},
 				DiffSuppressFunc: verify.SuppressMissingOptionalConfigurationBlock,
 			},
-
 			"name": {
-				Type:         schema.TypeString,
-				Required:     true,
-				ForceNew:     true,
-				ValidateFunc: validStateMachineName,
+				Type:          schema.TypeString,
+				Optional:      true,
+				Computed:      true,
+				ForceNew:      true,
+				ConflictsWith: []string{"name_prefix"},
+				ValidateFunc: validation.All(
+					validation.StringLenBetween(1, 80),
+					validation.StringMatch(regexp.MustCompile(`^[a-zA-Z0-9-_]+$`), "the name should only contain 0-9, A-Z, a-z, - and _"),
+				),
+			},
+			"name_prefix": {
+				Type:          schema.TypeString,
+				Optional:      true,
+				Computed:      true,
+				ForceNew:      true,
+				ConflictsWith: []string{"name"},
+				ValidateFunc: validation.All(
+					validation.StringLenBetween(1, 80-resource.UniqueIDSuffixLength),
+					validation.StringMatch(regexp.MustCompile(`^[a-zA-Z0-9-_]+$`), "the name should only contain 0-9, A-Z, a-z, - and _"),
+				),
 			},
-
 			"role_arn": {
 				Type:         schema.TypeString,
 				Required:     true,
 				ValidateFunc: verify.ValidARN,
 			},
-
 			"status": {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-
 			"tags":     tftags.TagsSchema(),
 			"tags_all": tftags.TagsSchemaComputed(),
-
-			"type": {
-				Type:         schema.TypeString,
-				Optional:     true,
-				ForceNew:     true,
-				Default:      sfn.StateMachineTypeStandard,
-				ValidateFunc: validation.StringInSlice(sfn.StateMachineType_Values(), false),
-			},
-
 			"tracing_configuration": {
 				Type:     schema.TypeList,
 				Optional: true,
@@ -113,18 +118,25 @@ func ResourceStateMachine() *schema.Resource {
 				},
 				DiffSuppressFunc: verify.SuppressMissingOptionalConfigurationBlock,
 			},
+			"type": {
+				Type:         schema.TypeString,
+				Optional:     true,
+				ForceNew:     true,
+				Default:      sfn.StateMachineTypeStandard,
+				ValidateFunc: validation.StringInSlice(sfn.StateMachineType_Values(), false),
+			},
 		},
 
 		CustomizeDiff: verify.SetTagsDiff,
 	}
 }
 
-func resourceStateMachineCreate(d *schema.ResourceData, meta interface{}) error {
+func resourceStateMachineCreate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
 	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
 	tags := defaultTagsConfig.MergeTags(tftags.New(d.Get("tags").(map[string]interface{})))
 
-	name := d.Get("name").(string)
+	name := create.Name(d.Get("name").(string), d.Get("name_prefix").(string))
 	input := &sfn.CreateStateMachineInput{
 		Definition: aws.String(d.Get("definition").(string)),
 		Name:       aws.String(name),
@@ -141,61 +153,38 @@ func resourceStateMachineCreate(d *schema.ResourceData, meta interface{}) error
 		input.TracingConfiguration = expandTracingConfiguration(v.([]interface{})[0].(map[string]interface{}))
 	}
 
-	var output *sfn.CreateStateMachineOutput
-
-	log.Printf("[DEBUG] Creating Step Function State Machine: %s", input)
-	err := resource.Retry(stateMachineCreatedTimeout, func() *resource.RetryError {
-		var err error
-
-		output, err = conn.CreateStateMachine(input)
-
-		// Note: the instance may be in a deleting mode, hence the retry
-		// when creating the step function. This can happen when we are
-		// updating the resource (since there is no update API call).
-		if tfawserr.ErrCodeEquals(err, sfn.ErrCodeStateMachineDeleting) {
-			return resource.RetryableError(err)
-		}
-
-		// This is done to deal with IAM eventual consistency
-		if tfawserr.ErrCodeEquals(err, "AccessDeniedException") {
-			return resource.RetryableError(err)
-		}
-
-		if err != nil {
-			return resource.NonRetryableError(err)
-		}
-
-		return nil
-	})
-
-	if tfresource.TimedOut(err) {
-		output, err = conn.CreateStateMachine(input)
-	}
+	// This is done to deal with IAM eventual consistency.
+	// Note: the instance may be in a deleting mode, hence the retry
+	// when creating the step function. This can happen when we are
+	// updating the resource (since there is no update API call).
+	outputRaw, err := tfresource.RetryWhenAWSErrCodeEqualsContext(ctx, stateMachineCreatedTimeout, func() (interface{}, error) {
+		return conn.CreateStateMachineWithContext(ctx, input)
+	}, sfn.ErrCodeStateMachineDeleting, "AccessDeniedException")
 
 	if err != nil {
-		return fmt.Errorf("error creating Step Function State Machine (%s): %w", name, err)
+		return diag.Errorf("creating Step Functions State Machine (%s): %s", name, err)
 	}
 
-	d.SetId(aws.StringValue(output.StateMachineArn))
+	d.SetId(aws.StringValue(outputRaw.(*sfn.CreateStateMachineOutput).StateMachineArn))
 
-	return resourceStateMachineRead(d, meta)
+	return resourceStateMachineRead(ctx, d, meta)
 }
 
-func resourceStateMachineRead(d *schema.ResourceData, meta interface{}) error {
+func resourceStateMachineRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
 	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
 	ignoreTagsConfig := meta.(*conns.AWSClient).IgnoreTagsConfig
 
-	output, err := FindStateMachineByARN(conn, d.Id())
+	output, err := FindStateMachineByARN(ctx, conn, d.Id())
 
 	if !d.IsNewResource() && tfresource.NotFound(err) {
-		log.Printf("[WARN] Step Function State Machine (%s) not found, removing from state", d.Id())
+		log.Printf("[WARN] Step Functions State Machine (%s) not found, removing from state", d.Id())
 		d.SetId("")
 		return nil
 	}
 
 	if err != nil {
-		return fmt.Errorf("error reading Step Function State Machine (%s): %w", d.Id(), err)
+		return diag.Errorf("reading Step Functions State Machine (%s): %s", d.Id(), err)
 	}
 
 	d.Set("arn", output.StateMachineArn)
@@ -205,60 +194,59 @@ func resourceStateMachineRead(d *schema.ResourceData, meta interface{}) error {
 		d.Set("creation_date", nil)
 	}
 	d.Set("definition", output.Definition)
-	d.Set("name", output.Name)
-	d.Set("role_arn", output.RoleArn)
-	d.Set("type", output.Type)
-	d.Set("status", output.Status)
-
 	if output.LoggingConfiguration != nil {
 		if err := d.Set("logging_configuration", []interface{}{flattenLoggingConfiguration(output.LoggingConfiguration)}); err != nil {
-			return fmt.Errorf("error setting logging_configuration: %w", err)
+			return diag.Errorf("setting logging_configuration: %s", err)
 		}
 	} else {
 		d.Set("logging_configuration", nil)
 	}
-
+	d.Set("name", output.Name)
+	d.Set("name_prefix", create.NamePrefixFromName(aws.StringValue(output.Name)))
+	d.Set("role_arn", output.RoleArn)
+	d.Set("status", output.Status)
 	if output.TracingConfiguration != nil {
 		if err := d.Set("tracing_configuration", []interface{}{flattenTracingConfiguration(output.TracingConfiguration)}); err != nil {
-			return fmt.Errorf("error setting tracing_configuration: %w", err)
+			return diag.Errorf("setting tracing_configuration: %s", err)
 		}
 	} else {
 		d.Set("tracing_configuration", nil)
 	}
+	d.Set("type", output.Type)
 
-	tags, err := ListTags(conn, d.Id())
+	tags, err := ListTagsWithContext(ctx, conn, d.Id())
 
-	if err != nil {
-		if tfawserr.ErrCodeEquals(err, "UnknownOperationException") {
-			return nil
-		}
+	if tfawserr.ErrCodeEquals(err, "UnknownOperationException") {
+		return nil
+	}
 
-		return fmt.Errorf("error listing tags for Step Function State Machine (%s): %w", d.Id(), err)
+	if err != nil {
+		return diag.Errorf("listing tags for Step Functions State Machine (%s): %s", d.Id(), err)
 	}
 
 	tags = 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)
+		return diag.Errorf("setting tags: %s", err)
 	}
 
 	if err := d.Set("tags_all", tags.Map()); err != nil {
-		return fmt.Errorf("error setting tags_all: %w", err)
+		return diag.Errorf("setting tags_all: %s", err)
 	}
 
 	return nil
 }
 
-func resourceStateMachineUpdate(d *schema.ResourceData, meta interface{}) error {
+func resourceStateMachineUpdate(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
 
 	if d.HasChangesExcept("tags", "tags_all") {
 		// "You must include at least one of definition or roleArn or you will receive a MissingRequiredParameter error"
 		input := &sfn.UpdateStateMachineInput{
-			StateMachineArn: aws.String(d.Id()),
 			Definition:      aws.String(d.Get("definition").(string)),
 			RoleArn:         aws.String(d.Get("role_arn").(string)),
+			StateMachineArn: aws.String(d.Id()),
 		}
 
 		if d.HasChange("logging_configuration") {
@@ -273,16 +261,15 @@ func resourceStateMachineUpdate(d *schema.ResourceData, meta interface{}) error
 			}
 		}
 
-		log.Printf("[DEBUG] Updating Step Function State Machine: %s", input)
-		_, err := conn.UpdateStateMachine(input)
+		_, err := conn.UpdateStateMachineWithContext(ctx, input)
 
 		if err != nil {
-			return fmt.Errorf("error updating Step Function State Machine (%s): %w", d.Id(), err)
+			return diag.Errorf("updating Step Functions State Machine (%s): %s", d.Id(), err)
 		}
 
 		// Handle eventual consistency after update.
-		err = resource.Retry(stateMachineUpdatedTimeout, func() *resource.RetryError {
-			output, err := FindStateMachineByARN(conn, d.Id())
+		err = resource.RetryContext(ctx, stateMachineUpdatedTimeout, func() *resource.RetryError { // nosemgrep:ci.helper-schema-resource-Retry-without-TimeoutError-check
+			output, err := FindStateMachineByARN(ctx, conn, d.Id())
 
 			if err != nil {
 				return resource.NonRetryableError(err)
@@ -293,45 +280,111 @@ func resourceStateMachineUpdate(d *schema.ResourceData, meta interface{}) error
 				d.HasChange("tracing_configuration.0.enabled") && output.TracingConfiguration != nil && aws.BoolValue(output.TracingConfiguration.Enabled) != d.Get("tracing_configuration.0.enabled").(bool) ||
 				d.HasChange("logging_configuration.0.include_execution_data") && output.LoggingConfiguration != nil && aws.BoolValue(output.LoggingConfiguration.IncludeExecutionData) != d.Get("logging_configuration.0.include_execution_data").(bool) ||
 				d.HasChange("logging_configuration.0.level") && output.LoggingConfiguration != nil && aws.StringValue(output.LoggingConfiguration.Level) != d.Get("logging_configuration.0.level").(string) {
-				return resource.RetryableError(fmt.Errorf("Step Function State Machine (%s) eventual consistency", d.Id()))
+				return resource.RetryableError(fmt.Errorf("Step Functions State Machine (%s) eventual consistency", d.Id()))
 			}
 
 			return nil
 		})
 
-		if tfresource.TimedOut(err) {
-			return fmt.Errorf("timed out waiting for Step Function State Machine (%s) update", d.Id())
+		if err != nil {
+			return diag.Errorf("waiting for Step Functions State Machine (%s) update: %s", d.Id(), err)
 		}
 	}
 
 	if d.HasChange("tags_all") {
 		o, n := d.GetChange("tags_all")
-		if err := UpdateTags(conn, d.Id(), o, n); err != nil {
-			return fmt.Errorf("error updating tags: %w", err)
+
+		if err := UpdateTagsWithContext(ctx, conn, d.Id(), o, n); err != nil {
+			return diag.Errorf("updating Step Functions State Machine (%s) tags: %s", d.Id(), err)
 		}
 	}
 
-	return resourceStateMachineRead(d, meta)
+	return resourceStateMachineRead(ctx, d, meta)
 }
 
-func resourceStateMachineDelete(d *schema.ResourceData, meta interface{}) error {
+func resourceStateMachineDelete(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
 
-	_, err := conn.DeleteStateMachine(&sfn.DeleteStateMachineInput{
+	log.Printf("[DEBUG] Deleting Step Functions State Machine: %s", d.Id())
+	_, err := conn.DeleteStateMachineWithContext(ctx, &sfn.DeleteStateMachineInput{
 		StateMachineArn: aws.String(d.Id()),
 	})
 
 	if err != nil {
-		return fmt.Errorf("error deleting Step Function State Machine (%s): %s", d.Id(), err)
+		return diag.Errorf("deleting Step Functions State Machine (%s): %s", d.Id(), err)
 	}
 
-	if _, err := waitStateMachineDeleted(conn, d.Id()); err != nil {
-		return fmt.Errorf("error waiting for Step Function State Machine (%s) deletion: %w", d.Id(), err)
+	if _, err := waitStateMachineDeleted(ctx, conn, d.Id()); err != nil {
+		return diag.Errorf("waiting for Step Functions State Machine (%s) delete: %s", d.Id(), err)
 	}
 
 	return nil
 }
 
+func FindStateMachineByARN(ctx context.Context, conn *sfn.SFN, arn string) (*sfn.DescribeStateMachineOutput, error) {
+	input := &sfn.DescribeStateMachineInput{
+		StateMachineArn: aws.String(arn),
+	}
+
+	output, err := conn.DescribeStateMachineWithContext(ctx, input)
+
+	if tfawserr.ErrCodeEquals(err, sfn.ErrCodeStateMachineDoesNotExist) {
+		return nil, &resource.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil {
+		return nil, tfresource.NewEmptyResultError(input)
+	}
+
+	return output, nil
+}
+
+func statusStateMachine(ctx context.Context, conn *sfn.SFN, stateMachineArn string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		output, err := FindStateMachineByARN(ctx, conn, stateMachineArn)
+
+		if tfresource.NotFound(err) {
+			return nil, "", nil
+		}
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		return output, aws.StringValue(output.Status), nil
+	}
+}
+
+const (
+	stateMachineCreatedTimeout = 5 * time.Minute
+	stateMachineDeletedTimeout = 5 * time.Minute
+	stateMachineUpdatedTimeout = 1 * time.Minute
+)
+
+func waitStateMachineDeleted(ctx context.Context, conn *sfn.SFN, stateMachineArn string) (*sfn.DescribeStateMachineOutput, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{sfn.StateMachineStatusActive, sfn.StateMachineStatusDeleting},
+		Target:  []string{},
+		Refresh: statusStateMachine(ctx, conn, stateMachineArn),
+		Timeout: stateMachineDeletedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForStateContext(ctx)
+
+	if output, ok := outputRaw.(*sfn.DescribeStateMachineOutput); ok {
+		return output, err
+	}
+
+	return nil, err
+}
+
 func expandLoggingConfiguration(tfMap map[string]interface{}) *sfn.LoggingConfiguration {
 	if tfMap == nil {
 		return nil
diff --git a/internal/service/sfn/state_machine_data_source.go b/internal/service/sfn/state_machine_data_source.go
index e461ef06003e..7597cf77ee64 100644
--- a/internal/service/sfn/state_machine_data_source.go
+++ b/internal/service/sfn/state_machine_data_source.go
@@ -1,29 +1,26 @@
 package sfn
 
 import (
-	"fmt"
-	"log"
+	"context"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/sfn"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
 )
 
 func DataSourceStateMachine() *schema.Resource {
 	return &schema.Resource{
-		Read: dataSourceStateMachineRead,
+		ReadWithoutTimeout: dataSourceStateMachineRead,
+
 		Schema: map[string]*schema.Schema{
-			"name": {
-				Type:     schema.TypeString,
-				Required: true,
-			},
 			"arn": {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-			"role_arn": {
+			"creation_date": {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
@@ -31,11 +28,15 @@ func DataSourceStateMachine() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-			"status": {
+			"name": {
+				Type:     schema.TypeString,
+				Required: true,
+			},
+			"role_arn": {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
-			"creation_date": {
+			"status": {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
@@ -43,51 +44,50 @@ func DataSourceStateMachine() *schema.Resource {
 	}
 }
 
-func dataSourceStateMachineRead(d *schema.ResourceData, meta interface{}) error {
+func dataSourceStateMachineRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	conn := meta.(*conns.AWSClient).SFNConn
-	params := &sfn.ListStateMachinesInput{}
-	log.Printf("[DEBUG] Reading Step Function State Machine: %s", d.Id())
 
-	target := d.Get("name")
+	name := d.Get("name").(string)
 	var arns []string
 
-	err := conn.ListStateMachinesPages(params, func(page *sfn.ListStateMachinesOutput, lastPage bool) bool {
-		for _, sm := range page.StateMachines {
-			if aws.StringValue(sm.Name) == target {
-				arns = append(arns, aws.StringValue(sm.StateMachineArn))
+	err := conn.ListStateMachinesPages(&sfn.ListStateMachinesInput{}, func(page *sfn.ListStateMachinesOutput, lastPage bool) bool {
+		if page == nil {
+			return !lastPage
+		}
+
+		for _, v := range page.StateMachines {
+			if aws.StringValue(v.Name) == name {
+				arns = append(arns, aws.StringValue(v.StateMachineArn))
 			}
 		}
-		return true
+
+		return !lastPage
 	})
 
 	if err != nil {
-		return fmt.Errorf("Error listing state machines: %w", err)
+		return diag.Errorf("listing Step Functions State Machines: %s", err)
 	}
 
-	if len(arns) == 0 {
-		return fmt.Errorf("No state machine with name %q found in this region.", target)
-	}
-	if len(arns) > 1 {
-		return fmt.Errorf("Multiple state machines with name %q found in this region.", target)
+	if n := len(arns); n == 0 {
+		return diag.Errorf("no Step Functions State Machines matched")
+	} else if n > 1 {
+		return diag.Errorf("%d Step Functions State Machines matched; use additional constraints to reduce matches to a single State Machine", n)
 	}
 
-	sm, err := conn.DescribeStateMachine(&sfn.DescribeStateMachineInput{
-		StateMachineArn: aws.String(arns[0]),
-	})
-	if err != nil {
-		return fmt.Errorf("error describing SFN State Machine (%s): %w", arns[0], err)
-	}
+	arn := arns[0]
+	output, err := FindStateMachineByARN(ctx, conn, arn)
 
-	d.Set("definition", sm.Definition)
-	d.Set("name", sm.Name)
-	d.Set("arn", sm.StateMachineArn)
-	d.Set("role_arn", sm.RoleArn)
-	d.Set("status", sm.Status)
-	if err := d.Set("creation_date", sm.CreationDate.Format(time.RFC3339)); err != nil {
-		log.Printf("[DEBUG] Error setting creation_date: %s", err)
+	if err != nil {
+		return diag.Errorf("reading Step Functions State Machine (%s): %s", arn, err)
 	}
 
-	d.SetId(arns[0])
+	d.SetId(arn)
+	d.Set("arn", output.StateMachineArn)
+	d.Set("creation_date", output.CreationDate.Format(time.RFC3339))
+	d.Set("definition", output.Definition)
+	d.Set("name", output.Name)
+	d.Set("role_arn", output.RoleArn)
+	d.Set("status", output.Status)
 
 	return nil
 }
diff --git a/internal/service/sfn/state_machine_data_source_test.go b/internal/service/sfn/state_machine_data_source_test.go
index e08cdd612f5b..a7e003765980 100644
--- a/internal/service/sfn/state_machine_data_source_test.go
+++ b/internal/service/sfn/state_machine_data_source_test.go
@@ -1,7 +1,6 @@
 package sfn_test
 
 import (
-	"fmt"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/service/sfn"
@@ -11,7 +10,7 @@ import (
 )
 
 func TestAccSFNStateMachineDataSource_basic(t *testing.T) {
-	rName := sdkacctest.RandString(5)
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
 	dataSourceName := "data.aws_sfn_state_machine.test"
 	resourceName := "aws_sfn_state_machine.test"
 
@@ -22,7 +21,7 @@ func TestAccSFNStateMachineDataSource_basic(t *testing.T) {
 		Steps: []resource.TestStep{
 			{
 				Config: testAccStateMachineDataSourceConfig_basic(rName),
-				Check: resource.ComposeTestCheckFunc(
+				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttrPair(resourceName, "id", dataSourceName, "arn"),
 					resource.TestCheckResourceAttrPair(resourceName, "creation_date", dataSourceName, "creation_date"),
 					resource.TestCheckResourceAttrPair(resourceName, "definition", dataSourceName, "definition"),
@@ -36,46 +35,9 @@ func TestAccSFNStateMachineDataSource_basic(t *testing.T) {
 }
 
 func testAccStateMachineDataSourceConfig_basic(rName string) string {
-	return fmt.Sprintf(`
-data "aws_region" "current" {}
-
-resource "aws_iam_role" "iam_for_sfn" {
-  name = "iam_for_sfn_%s"
-
-  assume_role_policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Effect": "Allow",
-      "Principal": {
-        "Service": "states.${data.aws_region.current.name}.amazonaws.com"
-      },
-      "Action": "sts:AssumeRole"
-    }
-  ]
-}
-EOF
-}
-
-resource "aws_sfn_state_machine" "test" {
-  name     = "test_sfn_%s"
-  role_arn = aws_iam_role.iam_for_sfn.arn
-
-  definition = <<EOF
-{
-  "StartAt": "HelloWorld",
-  "States": {
-    "HelloWorld": {
-      "Type": "Succeed"
-    }
-  }
-}
-EOF
-}
-
+	return acctest.ConfigCompose(testAccStateMachineConfig_basic(rName, 5), `
 data "aws_sfn_state_machine" "test" {
   name = aws_sfn_state_machine.test.name
 }
-`, rName, rName)
+`)
 }
diff --git a/internal/service/sfn/state_machine_test.go b/internal/service/sfn/state_machine_test.go
index f1b81077a197..cc297e9238e9 100644
--- a/internal/service/sfn/state_machine_test.go
+++ b/internal/service/sfn/state_machine_test.go
@@ -1,6 +1,7 @@
 package sfn_test
 
 import (
+	"context"
 	"fmt"
 	"regexp"
 	"testing"
@@ -34,6 +35,7 @@ func TestAccSFNStateMachine_createUpdate(t *testing.T) {
 					acctest.CheckResourceAttrRegionalARN(resourceName, "arn", "states", fmt.Sprintf("stateMachine:%s", rName)),
 					resource.TestCheckResourceAttr(resourceName, "status", sfn.StateMachineStatusActive),
 					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestCheckResourceAttr(resourceName, "name_prefix", ""),
 					resource.TestCheckResourceAttrSet(resourceName, "creation_date"),
 					resource.TestCheckResourceAttrSet(resourceName, "definition"),
 					resource.TestMatchResourceAttr(resourceName, "definition", regexp.MustCompile(`.*\"MaxAttempts\": 5.*`)),
@@ -157,6 +159,62 @@ func TestAccSFNStateMachine_standardUpdate(t *testing.T) {
 	})
 }
 
+func TestAccSFNStateMachine_nameGenerated(t *testing.T) {
+	var sm sfn.DescribeStateMachineOutput
+	resourceName := "aws_sfn_state_machine.test"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(t) },
+		ErrorCheck:               acctest.ErrorCheck(t, sfn.EndpointsID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckStateMachineDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStateMachineConfig_nameGenerated(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckExists(resourceName, &sm),
+					acctest.CheckResourceAttrNameGenerated(resourceName, "name"),
+					resource.TestCheckResourceAttr(resourceName, "name_prefix", resource.UniqueIdPrefix),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
+func TestAccSFNStateMachine_namePrefix(t *testing.T) {
+	var sm sfn.DescribeStateMachineOutput
+	resourceName := "aws_sfn_state_machine.test"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(t) },
+		ErrorCheck:               acctest.ErrorCheck(t, sfn.EndpointsID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		CheckDestroy:             testAccCheckStateMachineDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStateMachineConfig_namePrefix(rName, "tf-acc-test-prefix-"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckExists(resourceName, &sm),
+					acctest.CheckResourceAttrNameFromPrefix(resourceName, "name", "tf-acc-test-prefix-"),
+					resource.TestCheckResourceAttr(resourceName, "name_prefix", "tf-acc-test-prefix-"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func TestAccSFNStateMachine_tags(t *testing.T) {
 	var sm sfn.DescribeStateMachineOutput
 	resourceName := "aws_sfn_state_machine.test"
@@ -311,12 +369,12 @@ func testAccCheckExists(n string, v *sfn.DescribeStateMachineOutput) resource.Te
 		}
 
 		if rs.Primary.ID == "" {
-			return fmt.Errorf("No Step Function State Machine ID is set")
+			return fmt.Errorf("No Step Functions State Machine ID is set")
 		}
 
 		conn := acctest.Provider.Meta().(*conns.AWSClient).SFNConn
 
-		output, err := tfsfn.FindStateMachineByARN(conn, rs.Primary.ID)
+		output, err := tfsfn.FindStateMachineByARN(context.Background(), conn, rs.Primary.ID)
 
 		if err != nil {
 			return err
@@ -336,7 +394,7 @@ func testAccCheckStateMachineDestroy(s *terraform.State) error {
 			continue
 		}
 
-		_, err := tfsfn.FindStateMachineByARN(conn, rs.Primary.ID)
+		_, err := tfsfn.FindStateMachineByARN(context.Background(), conn, rs.Primary.ID)
 
 		if tfresource.NotFound(err) {
 			continue
@@ -346,13 +404,13 @@ func testAccCheckStateMachineDestroy(s *terraform.State) error {
 			return err
 		}
 
-		return fmt.Errorf("Step Function State Machine %s still exists", rs.Primary.ID)
+		return fmt.Errorf("Step Functions State Machine %s still exists", rs.Primary.ID)
 	}
 
 	return nil
 }
 
-func testAccStateMachineBaseConfig(rName string) string {
+func testAccStateMachineConfig_base(rName string) string {
 	return fmt.Sprintf(`
 resource "aws_iam_role_policy" "for_lambda" {
   name = "%[1]s-lambda"
@@ -453,7 +511,7 @@ EOF
 }
 
 func testAccStateMachineConfig_basic(rName string, rMaxAttempts int) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_sfn_state_machine" "test" {
   name     = %[1]q
   role_arn = aws_iam_role.for_sfn.arn
@@ -485,8 +543,73 @@ EOF
 `, rName, rMaxAttempts))
 }
 
+func testAccStateMachineConfig_nameGenerated(rName string) string {
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), `
+resource "aws_sfn_state_machine" "test" {
+  role_arn = aws_iam_role.for_sfn.arn
+
+  definition = <<EOF
+{
+  "Comment": "A Hello World example of the Amazon States Language using an AWS Lambda Function",
+  "StartAt": "HelloWorld",
+  "States": {
+    "HelloWorld": {
+      "Type": "Task",
+      "Resource": "${aws_lambda_function.test.arn}",
+      "Retry": [
+        {
+          "ErrorEquals": [
+            "States.ALL"
+          ],
+          "IntervalSeconds": 5,
+          "MaxAttempts": 5,
+          "BackoffRate": 8
+        }
+      ],
+      "End": true
+    }
+  }
+}
+EOF
+}
+`)
+}
+
+func testAccStateMachineConfig_namePrefix(rName, namePrefix string) string {
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
+resource "aws_sfn_state_machine" "test" {
+  name_prefix = %[1]q
+  role_arn    = aws_iam_role.for_sfn.arn
+
+  definition = <<EOF
+{
+  "Comment": "A Hello World example of the Amazon States Language using an AWS Lambda Function",
+  "StartAt": "HelloWorld",
+  "States": {
+    "HelloWorld": {
+      "Type": "Task",
+      "Resource": "${aws_lambda_function.test.arn}",
+      "Retry": [
+        {
+          "ErrorEquals": [
+            "States.ALL"
+          ],
+          "IntervalSeconds": 5,
+          "MaxAttempts": 5,
+          "BackoffRate": 8
+        }
+      ],
+      "End": true
+    }
+  }
+}
+EOF
+}
+`, namePrefix))
+}
+
 func testAccStateMachineConfig_tags1(rName, tag1Key, tag1Value string) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_sfn_state_machine" "test" {
   name     = %[1]q
   role_arn = aws_iam_role.for_sfn.arn
@@ -523,7 +646,7 @@ EOF
 }
 
 func testAccStateMachineConfig_tags2(rName, tag1Key, tag1Value, tag2Key, tag2Value string) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_sfn_state_machine" "test" {
   name     = %[1]q
   role_arn = aws_iam_role.for_sfn.arn
@@ -561,7 +684,7 @@ EOF
 }
 
 func testAccStateMachineConfig_typed(rName, rType string, rMaxAttempts int) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_sfn_state_machine" "test" {
   name     = %[1]q
   role_arn = aws_iam_role.for_sfn.arn
@@ -593,7 +716,7 @@ EOF
 }
 
 func testAccStateMachineConfig_expressLogConfiguration(rName string, rLevel string) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_cloudwatch_log_group" "test" {
   name = %[1]q
 }
@@ -637,7 +760,7 @@ EOF
 }
 
 func testAccStateMachineConfig_tracingEnable(rName string) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_sfn_state_machine" "test" {
   name     = %[1]q
   role_arn = aws_iam_role.for_sfn.arn
@@ -674,7 +797,7 @@ EOF
 }
 
 func testAccStateMachineConfig_tracingDisable(rName string) string {
-	return acctest.ConfigCompose(testAccStateMachineBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccStateMachineConfig_base(rName), fmt.Sprintf(`
 resource "aws_sfn_state_machine" "test" {
   name     = %[1]q
   role_arn = aws_iam_role.for_sfn.arn
diff --git a/internal/service/sfn/status.go b/internal/service/sfn/status.go
deleted file mode 100644
index be16920c4357..000000000000
--- a/internal/service/sfn/status.go
+++ /dev/null
@@ -1,24 +0,0 @@
-package sfn
-
-import (
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/sfn"
-	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
-	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
-)
-
-func statusStateMachine(conn *sfn.SFN, stateMachineArn string) resource.StateRefreshFunc {
-	return func() (interface{}, string, error) {
-		output, err := FindStateMachineByARN(conn, stateMachineArn)
-
-		if tfresource.NotFound(err) {
-			return nil, "", nil
-		}
-
-		if err != nil {
-			return nil, "", err
-		}
-
-		return output, aws.StringValue(output.Status), nil
-	}
-}
diff --git a/internal/service/sfn/sweep.go b/internal/service/sfn/sweep.go
new file mode 100644
index 000000000000..5eaf8e148bd1
--- /dev/null
+++ b/internal/service/sfn/sweep.go
@@ -0,0 +1,113 @@
+//go:build sweep
+// +build sweep
+
+package sfn
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/sfn"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-provider-aws/internal/conns"
+	"github.com/hashicorp/terraform-provider-aws/internal/sweep"
+)
+
+func init() {
+	resource.AddTestSweepers("aws_sfn_activity", &resource.Sweeper{
+		Name: "aws_sfn_activity",
+		F:    sweepActivities,
+	})
+
+	resource.AddTestSweepers("aws_sfn_state_machine", &resource.Sweeper{
+		Name: "aws_sfn_state_machine",
+		F:    sweepStateMachines,
+	})
+}
+
+func sweepActivities(region string) error {
+	client, err := sweep.SharedRegionalSweepClient(region)
+	if err != nil {
+		return fmt.Errorf("error getting client: %s", err)
+	}
+	conn := client.(*conns.AWSClient).SFNConn
+	input := &sfn.ListActivitiesInput{}
+	sweepResources := make([]sweep.Sweepable, 0)
+
+	err = conn.ListActivitiesPages(input, func(page *sfn.ListActivitiesOutput, lastPage bool) bool {
+		if page == nil {
+			return !lastPage
+		}
+
+		for _, v := range page.Activities {
+			r := ResourceActivity()
+			d := r.Data(nil)
+			d.SetId(aws.StringValue(v.ActivityArn))
+
+			sweepResources = append(sweepResources, sweep.NewSweepResource(r, d, client))
+		}
+
+		return !lastPage
+	})
+
+	if sweep.SkipSweepError(err) {
+		log.Printf("[WARN] Skipping Step Functions Activity sweep for %s: %s", region, err)
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("error listing Step Functions Activities (%s): %w", region, err)
+	}
+
+	err = sweep.SweepOrchestrator(sweepResources)
+
+	if err != nil {
+		return fmt.Errorf("error sweeping Step Functions Activities (%s): %w", region, err)
+	}
+
+	return nil
+}
+
+func sweepStateMachines(region string) error {
+	client, err := sweep.SharedRegionalSweepClient(region)
+	if err != nil {
+		return fmt.Errorf("error getting client: %s", err)
+	}
+	conn := client.(*conns.AWSClient).SFNConn
+	input := &sfn.ListStateMachinesInput{}
+	sweepResources := make([]sweep.Sweepable, 0)
+
+	err = conn.ListStateMachinesPages(input, func(page *sfn.ListStateMachinesOutput, lastPage bool) bool {
+		if page == nil {
+			return !lastPage
+		}
+
+		for _, v := range page.StateMachines {
+			r := ResourceStateMachine()
+			d := r.Data(nil)
+			d.SetId(aws.StringValue(v.StateMachineArn))
+
+			sweepResources = append(sweepResources, sweep.NewSweepResource(r, d, client))
+		}
+
+		return !lastPage
+	})
+
+	if sweep.SkipSweepError(err) {
+		log.Printf("[WARN] Skipping Step Functions State Machine sweep for %s: %s", region, err)
+		return nil
+	}
+
+	if err != nil {
+		return fmt.Errorf("error listing Step Functions State Machines (%s): %w", region, err)
+	}
+
+	err = sweep.SweepOrchestrator(sweepResources)
+
+	if err != nil {
+		return fmt.Errorf("error sweeping Step Functions State Machines (%s): %w", region, err)
+	}
+
+	return nil
+}
diff --git a/internal/service/sfn/validate.go b/internal/service/sfn/validate.go
deleted file mode 100644
index a0f860ec8df3..000000000000
--- a/internal/service/sfn/validate.go
+++ /dev/null
@@ -1,19 +0,0 @@
-package sfn
-
-import (
-	"fmt"
-	"regexp"
-)
-
-func validStateMachineName(v interface{}, k string) (ws []string, errors []error) {
-	value := v.(string)
-	if len(value) > 80 {
-		errors = append(errors, fmt.Errorf("%q cannot be longer than 80 characters", k))
-	}
-
-	if !regexp.MustCompile(`^[a-zA-Z0-9-_]+$`).MatchString(value) {
-		errors = append(errors, fmt.Errorf(
-			"%q must be composed with only these characters [a-zA-Z0-9-_]: %v", k, value))
-	}
-	return
-}
diff --git a/internal/service/sfn/validate_test.go b/internal/service/sfn/validate_test.go
deleted file mode 100644
index a00f4492153b..000000000000
--- a/internal/service/sfn/validate_test.go
+++ /dev/null
@@ -1,51 +0,0 @@
-package sfn
-
-import (
-	"strings"
-	"testing"
-)
-
-func TestValidStateMachineName(t *testing.T) {
-	validTypes := []string{
-		"foo",
-		"BAR",
-		"FooBar123",
-		"FooBar123Baz-_",
-	}
-
-	invalidTypes := []string{
-		"foo bar",
-		"foo<bar>",
-		"foo{bar}",
-		"foo[bar]",
-		"foo*bar",
-		"foo?bar",
-		"foo#bar",
-		"foo%bar",
-		"foo\bar",
-		"foo^bar",
-		"foo|bar",
-		"foo~bar",
-		"foo$bar",
-		"foo&bar",
-		"foo,bar",
-		"foo:bar",
-		"foo;bar",
-		"foo/bar",
-		strings.Repeat("W", 81), // length > 80
-	}
-
-	for _, v := range validTypes {
-		_, errors := validStateMachineName(v, "name")
-		if len(errors) != 0 {
-			t.Fatalf("%q should be a valid Step Function State Machine name: %v", v, errors)
-		}
-	}
-
-	for _, v := range invalidTypes {
-		_, errors := validStateMachineName(v, "name")
-		if len(errors) == 0 {
-			t.Fatalf("%q should not be a valid Step Function State Machine name", v)
-		}
-	}
-}
diff --git a/internal/service/sfn/wait.go b/internal/service/sfn/wait.go
deleted file mode 100644
index df407238cd83..000000000000
--- a/internal/service/sfn/wait.go
+++ /dev/null
@@ -1,31 +0,0 @@
-package sfn
-
-import (
-	"time"
-
-	"github.com/aws/aws-sdk-go/service/sfn"
-	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
-)
-
-const (
-	stateMachineCreatedTimeout = 5 * time.Minute
-	stateMachineDeletedTimeout = 5 * time.Minute
-	stateMachineUpdatedTimeout = 1 * time.Minute
-)
-
-func waitStateMachineDeleted(conn *sfn.SFN, stateMachineArn string) (*sfn.DescribeStateMachineOutput, error) {
-	stateConf := &resource.StateChangeConf{
-		Pending: []string{sfn.StateMachineStatusActive, sfn.StateMachineStatusDeleting},
-		Target:  []string{},
-		Refresh: statusStateMachine(conn, stateMachineArn),
-		Timeout: stateMachineDeletedTimeout,
-	}
-
-	outputRaw, err := stateConf.WaitForState()
-
-	if output, ok := outputRaw.(*sfn.DescribeStateMachineOutput); ok {
-		return output, err
-	}
-
-	return nil, err
-}
diff --git a/internal/service/swf/domain.go b/internal/service/swf/domain.go
index 1a3705df7879..ec289c101b31 100644
--- a/internal/service/swf/domain.go
+++ b/internal/service/swf/domain.go
@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"log"
 	"strconv"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/swf"
@@ -176,14 +175,6 @@ func resourceDomainDelete(ctx context.Context, d *schema.ResourceData, meta inte
 		return diag.Errorf("deleting SWF Domain (%s): %s", d.Id(), err)
 	}
 
-	_, err = tfresource.RetryUntilNotFoundContext(ctx, 1*time.Minute, func() (interface{}, error) {
-		return FindDomainByName(ctx, conn, d.Id())
-	})
-
-	if err != nil {
-		return diag.Errorf("waiting for SWF Domain (%s) delete: %s", d.Id(), err)
-	}
-
 	return nil
 }
 
diff --git a/internal/service/swf/domain_test.go b/internal/service/swf/domain_test.go
index 45cf9175ea1b..05f5b30ca503 100644
--- a/internal/service/swf/domain_test.go
+++ b/internal/service/swf/domain_test.go
@@ -6,6 +6,7 @@ import (
 	"os"
 	"regexp"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/service/swf"
 	sdkacctest "github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
@@ -202,17 +203,22 @@ func testAccCheckDomainDestroy(s *terraform.State) error {
 			continue
 		}
 
-		_, err := tfswf.FindDomainByName(context.Background(), conn, rs.Primary.ID)
+		// Retrying as Read after Delete is not always consistent.
+		err := resource.Retry(2*time.Minute, func() *resource.RetryError {
+			_, err := tfswf.FindDomainByName(context.Background(), conn, rs.Primary.ID)
 
-		if tfresource.NotFound(err) {
-			continue
-		}
+			if tfresource.NotFound(err) {
+				return nil
+			}
 
-		if err != nil {
-			return err
-		}
+			if err != nil {
+				return resource.NonRetryableError(err)
+			}
+
+			return resource.RetryableError(fmt.Errorf("SWF Domain still exists: %s", rs.Primary.ID))
+		})
 
-		return fmt.Errorf("SWF Domain still exists: %s", rs.Primary.ID)
+		return err
 	}
 
 	return nil
diff --git a/internal/sweep/sweep_test.go b/internal/sweep/sweep_test.go
index 173e2df796fb..2df023824f47 100644
--- a/internal/sweep/sweep_test.go
+++ b/internal/sweep/sweep_test.go
@@ -122,6 +122,7 @@ import (
 	_ "github.com/hashicorp/terraform-provider-aws/internal/service/servicecatalog"
 	_ "github.com/hashicorp/terraform-provider-aws/internal/service/servicediscovery"
 	_ "github.com/hashicorp/terraform-provider-aws/internal/service/ses"
+	_ "github.com/hashicorp/terraform-provider-aws/internal/service/sfn"
 	_ "github.com/hashicorp/terraform-provider-aws/internal/service/simpledb"
 	_ "github.com/hashicorp/terraform-provider-aws/internal/service/sns"
 	_ "github.com/hashicorp/terraform-provider-aws/internal/service/sqs"
diff --git a/website/docs/r/sfn_state_machine.html.markdown b/website/docs/r/sfn_state_machine.html.markdown
index c001979d5331..cfd7404757f8 100644
--- a/website/docs/r/sfn_state_machine.html.markdown
+++ b/website/docs/r/sfn_state_machine.html.markdown
@@ -101,7 +101,8 @@ The following arguments are supported:
 
 * `definition` - (Required) The [Amazon States Language](https://docs.aws.amazon.com/step-functions/latest/dg/concepts-amazon-states-language.html) definition of the state machine.
 * `logging_configuration` - (Optional) Defines what execution history events are logged and where they are logged. The `logging_configuration` parameter is only valid when `type` is set to `EXPRESS`. Defaults to `OFF`. For more information see [Logging Express Workflows](https://docs.aws.amazon.com/step-functions/latest/dg/cw-logs.html) and [Log Levels](https://docs.aws.amazon.com/step-functions/latest/dg/cloudwatch-log-level.html) in the AWS Step Functions User Guide.
-* `name` - (Required) The name of the state machine. To enable logging with CloudWatch Logs, the name should only contain `0`-`9`, `A`-`Z`, `a`-`z`, `-` and `_`.
+* `name` - (Optional) The name of the state machine. The name should only contain `0`-`9`, `A`-`Z`, `a`-`z`, `-` and `_`. If omitted, Terraform will assign a random, unique name.
+* `name_prefix` - (Optional) Creates a unique name beginning with the specified prefix. Conflicts with `name`.
 * `role_arn` - (Required) The Amazon Resource Name (ARN) of the IAM role to use for this state machine.
 * `tags` - (Optional) Key-value map of resource tags. If configured with a provider [`default_tags` configuration block](https://registry.terraform.io/providers/hashicorp/aws/latest/docs#default_tags-configuration-block) present, tags with matching keys will overwrite those defined at the provider-level.
 * `tracing_configuration` - (Optional) Selects whether AWS X-Ray tracing is enabled.