diff --git a/aws/resource_aws_sfn_state_machine.go b/aws/resource_aws_sfn_state_machine.go
index bbeb5ed6e34..81825478563 100644
--- a/aws/resource_aws_sfn_state_machine.go
+++ b/aws/resource_aws_sfn_state_machine.go
@@ -6,7 +6,6 @@ import (
 	"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/terraform-plugin-sdk/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
@@ -54,6 +53,16 @@ func resourceAwsSfnStateMachine() *schema.Resource {
 				Computed: true,
 			},
 			"tags": tagsSchema(),
+			"type": {
+				Type:     schema.TypeString,
+				Optional: true,
+				ForceNew: true,
+				Default:  sfn.StateMachineTypeStandard,
+				ValidateFunc: validation.StringInSlice([]string{
+					sfn.StateMachineTypeStandard,
+					sfn.StateMachineTypeExpress,
+				}, false),
+			},
 		},
 	}
 }
@@ -66,20 +75,25 @@ func resourceAwsSfnStateMachineCreate(d *schema.ResourceData, meta interface{})
 		Definition: aws.String(d.Get("definition").(string)),
 		Name:       aws.String(d.Get("name").(string)),
 		RoleArn:    aws.String(d.Get("role_arn").(string)),
+		Type:       aws.String(d.Get("type").(string)),
 		Tags:       keyvaluetags.New(d.Get("tags").(map[string]interface{})).IgnoreAws().SfnTags(),
 	}
 
-	var activity *sfn.CreateStateMachineOutput
+	var stateMachine *sfn.CreateStateMachineOutput
 
 	err := resource.Retry(5*time.Minute, func() *resource.RetryError {
 		var err error
-		activity, err = conn.CreateStateMachine(params)
+		stateMachine, err = conn.CreateStateMachine(params)
 
 		if err != nil {
 			// 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 awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == "StateMachineDeleting" {
+			if isAWSErr(err, sfn.ErrCodeStateMachineDeleting, "") {
+				return resource.RetryableError(err)
+			}
+			//This is done to deal with IAM eventual consistency
+			if isAWSErr(err, "AccessDeniedException", "") {
 				return resource.RetryableError(err)
 			}
 
@@ -89,14 +103,14 @@ func resourceAwsSfnStateMachineCreate(d *schema.ResourceData, meta interface{})
 		return nil
 	})
 	if isResourceTimeoutError(err) {
-		activity, err = conn.CreateStateMachine(params)
+		stateMachine, err = conn.CreateStateMachine(params)
 	}
 
 	if err != nil {
 		return fmt.Errorf("Error creating Step Function State Machine: %s", err)
 	}
 
-	d.SetId(*activity.StateMachineArn)
+	d.SetId(*stateMachine.StateMachineArn)
 
 	return resourceAwsSfnStateMachineRead(d, meta)
 }
@@ -110,11 +124,9 @@ func resourceAwsSfnStateMachineRead(d *schema.ResourceData, meta interface{}) er
 	})
 	if err != nil {
 
-		if awserr, ok := err.(awserr.Error); ok {
-			if awserr.Code() == "NotFoundException" || awserr.Code() == "StateMachineDoesNotExist" {
-				d.SetId("")
-				return nil
-			}
+		if isAWSErr(err, sfn.ErrCodeStateMachineDoesNotExist, "") {
+			d.SetId("")
+			return nil
 		}
 		return err
 	}
@@ -123,6 +135,7 @@ func resourceAwsSfnStateMachineRead(d *schema.ResourceData, meta interface{}) er
 	d.Set("name", sm.Name)
 	d.Set("role_arn", sm.RoleArn)
 	d.Set("status", sm.Status)
+	d.Set("type", sm.Type)
 
 	if err := d.Set("creation_date", sm.CreationDate.Format(time.RFC3339)); err != nil {
 		log.Printf("[DEBUG] Error setting creation_date: %s", err)
@@ -155,7 +168,7 @@ func resourceAwsSfnStateMachineUpdate(d *schema.ResourceData, meta interface{})
 	log.Printf("[DEBUG] Updating Step Function State Machine: %#v", params)
 
 	if err != nil {
-		if isAWSErr(err, "StateMachineDoesNotExist", "State Machine Does Not Exist") {
+		if isAWSErr(err, sfn.ErrCodeStateMachineDoesNotExist, "State Machine Does Not Exist") {
 			return fmt.Errorf("Error updating Step Function State Machine: %s", err)
 		}
 		return err
diff --git a/aws/resource_aws_sfn_state_machine_test.go b/aws/resource_aws_sfn_state_machine_test.go
index beec17c2dbd..9b2c53c1df2 100644
--- a/aws/resource_aws_sfn_state_machine_test.go
+++ b/aws/resource_aws_sfn_state_machine_test.go
@@ -6,7 +6,6 @@ import (
 	"testing"
 
 	"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/terraform-plugin-sdk/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
@@ -14,7 +13,8 @@ import (
 )
 
 func TestAccAWSSfnStateMachine_createUpdate(t *testing.T) {
-	name := acctest.RandString(10)
+	resourceName := "aws_sfn_state_machine.test"
+	rName := acctest.RandomWithPrefix("tf-acc")
 
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -22,34 +22,74 @@ func TestAccAWSSfnStateMachine_createUpdate(t *testing.T) {
 		CheckDestroy: testAccCheckAWSSfnStateMachineDestroy,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccAWSSfnStateMachineConfig(name, 5),
+				Config: testAccAWSSfnStateMachineConfig(rName, 5),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSSfnExists("aws_sfn_state_machine.foo"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "status", sfn.StateMachineStatusActive),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "name"),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "creation_date"),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "definition"),
-					resource.TestMatchResourceAttr("aws_sfn_state_machine.foo", "definition", regexp.MustCompile(`.*\"MaxAttempts\": 5.*`)),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "role_arn"),
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "status", sfn.StateMachineStatusActive),
+					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestCheckResourceAttr(resourceName, "type", sfn.StateMachineTypeStandard),
+					resource.TestCheckResourceAttrSet(resourceName, "creation_date"),
+					resource.TestCheckResourceAttrSet(resourceName, "definition"),
+					resource.TestMatchResourceAttr(resourceName, "definition", regexp.MustCompile(`.*\"MaxAttempts\": 5.*`)),
+					testAccMatchResourceAttrGlobalARN(resourceName, "role_arn", "iam", regexp.MustCompile(`role/.+`)),
 				),
 			},
 			{
-				Config: testAccAWSSfnStateMachineConfig(name, 10),
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccAWSSfnStateMachineConfig(rName, 10),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSSfnExists("aws_sfn_state_machine.foo"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "status", sfn.StateMachineStatusActive),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "name"),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "creation_date"),
-					resource.TestMatchResourceAttr("aws_sfn_state_machine.foo", "definition", regexp.MustCompile(`.*\"MaxAttempts\": 10.*`)),
-					resource.TestCheckResourceAttrSet("aws_sfn_state_machine.foo", "role_arn"),
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "status", sfn.StateMachineStatusActive),
+					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestCheckResourceAttr(resourceName, "type", sfn.StateMachineTypeStandard),
+					resource.TestCheckResourceAttrSet(resourceName, "creation_date"),
+					resource.TestMatchResourceAttr(resourceName, "definition", regexp.MustCompile(`.*\"MaxAttempts\": 10.*`)),
+					testAccMatchResourceAttrGlobalARN(resourceName, "role_arn", "iam", regexp.MustCompile(`role/.+`)),
 				),
 			},
 		},
 	})
 }
 
+func TestAccAWSSfnStateMachine_type_express(t *testing.T) {
+	resourceName := "aws_sfn_state_machine.test"
+	rName := acctest.RandomWithPrefix("tf-acc")
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSSfnStateMachineDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccAWSSfnStateMachineConfigType(rName, "EXPRESS"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestCheckResourceAttr(resourceName, "type", sfn.StateMachineTypeExpress)),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccAWSSfnStateMachineConfigType(rName, "STANDARD"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "name", rName),
+					resource.TestCheckResourceAttr(resourceName, "type", sfn.StateMachineTypeStandard)),
+			},
+		},
+	})
+}
+
 func TestAccAWSSfnStateMachine_Tags(t *testing.T) {
-	name := acctest.RandString(10)
+	resourceName := "aws_sfn_state_machine.test"
+	rName := acctest.RandomWithPrefix("tf-acc")
 
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:     func() { testAccPreCheck(t) },
@@ -57,30 +97,60 @@ func TestAccAWSSfnStateMachine_Tags(t *testing.T) {
 		CheckDestroy: testAccCheckAWSSfnStateMachineDestroy,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccAWSSfnStateMachineConfigTags1(name, "key1", "value1"),
+				Config: testAccAWSSfnStateMachineConfigTags1(rName, "key1", "value1"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key1", "value1"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccAWSSfnStateMachineConfigTags2(rName, "key1", "value1updated", "key2", "value2"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSSfnExists("aws_sfn_state_machine.foo"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.%", "1"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.key1", "value1"),
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "2"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key1", "value1updated"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key2", "value2"),
 				),
 			},
 			{
-				Config: testAccAWSSfnStateMachineConfigTags2(name, "key1", "value1updated", "key2", "value2"),
+				Config: testAccAWSSfnStateMachineConfigTags1(rName, "key2", "value2"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSSfnExists("aws_sfn_state_machine.foo"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.%", "2"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.key1", "value1updated"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.key2", "value2"),
+					testAccCheckAWSSfnExists(resourceName),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key2", "value2"),
 				),
 			},
+		},
+	})
+}
+
+func TestAccAWSSfnStateMachine_disappears(t *testing.T) {
+	var sm sfn.DescribeStateMachineOutput
+
+	resourceName := "aws_sfn_state_machine.test"
+	rName := acctest.RandomWithPrefix("tf-acc")
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckAWSSfnStateMachineDestroy,
+		Steps: []resource.TestStep{
 			{
-				Config: testAccAWSSfnStateMachineConfigTags1(name, "key2", "value2"),
+				Config: testAccAWSSfnStateMachineConfig(rName, 5),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckAWSSfnExists("aws_sfn_state_machine.foo"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.%", "1"),
-					resource.TestCheckResourceAttr("aws_sfn_state_machine.foo", "tags.key2", "value2"),
+					testAccCheckAWSSfnExists(resourceName),
+					testAccCheckAWSSfnStateMachineDisappears(&sm),
 				),
 			},
+			{
+				ExpectNonEmptyPlan: true,
+			},
 		},
 	})
 }
@@ -119,7 +189,7 @@ func testAccCheckAWSSfnStateMachineDestroy(s *terraform.State) error {
 		})
 
 		if err != nil {
-			if wserr, ok := err.(awserr.Error); ok && wserr.Code() == "StateMachineDoesNotExist" {
+			if isAWSErr(err, sfn.ErrCodeStateMachineDoesNotExist, "") {
 				return nil
 			}
 			return err
@@ -135,32 +205,22 @@ func testAccCheckAWSSfnStateMachineDestroy(s *terraform.State) error {
 	return fmt.Errorf("Default error in Step Function Test")
 }
 
-func testAccAWSSfnStateMachineConfig(rName string, rMaxAttempts int) string {
-	return fmt.Sprintf(`
-data "aws_partition" "current" {}
+func testAccCheckAWSSfnStateMachineDisappears(sm *sfn.DescribeStateMachineOutput) resource.TestCheckFunc {
+	return func(s *terraform.State) error {
+		conn := testAccProvider.Meta().(*AWSClient).sfnconn
 
-data "aws_region" "current" {}
+		input := &sfn.DeleteStateMachineInput{
+			StateMachineArn: sm.StateMachineArn,
+		}
 
-resource "aws_iam_role_policy" "iam_policy_for_lambda" {
-  name = "iam_policy_for_lambda_%s"
-  role = "${aws_iam_role.iam_for_lambda.id}"
+		_, err := conn.DeleteStateMachine(input)
 
-  policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [{
-    "Effect": "Allow",
-    "Action": [
-      "logs:CreateLogGroup",
-      "logs:CreateLogStream",
-      "logs:PutLogEvents"
-    ],
-    "Resource": "arn:${data.aws_partition.current.partition}:logs:*:*:*"
-  }]
-}
-EOF
+		return err
+	}
 }
 
+func testAccAWSSfnStateMachineConfigBase(rName string) string {
+	return fmt.Sprintf(`
 resource "aws_iam_role" "iam_for_lambda" {
   name = "iam_for_lambda_%s"
 
@@ -181,8 +241,18 @@ resource "aws_iam_role" "iam_for_lambda" {
 EOF
 }
 
+resource "aws_lambda_function" "test" {
+  filename      = "test-fixtures/lambdatest.zip"
+  function_name = %[1]q
+  role          = "${aws_iam_role.iam_for_lambda.arn}"
+  handler       = "exports.example"
+  runtime       = "nodejs12.x"
+}
+
+data "aws_region" "current" {}
+
 resource "aws_iam_role_policy" "iam_policy_for_sfn" {
-  name = "iam_policy_for_sfn_%s"
+  name = %[1]q
   role = "${aws_iam_role.iam_for_sfn.id}"
 
   policy = <<EOF
@@ -194,15 +264,15 @@ resource "aws_iam_role_policy" "iam_policy_for_sfn" {
       "Action": [
         "lambda:InvokeFunction"
       ],
-        "Resource": "*"
-      }
+        "Resource": "${aws_lambda_function.test.arn}"
+	}
   ]
 }
 EOF
 }
 
 resource "aws_iam_role" "iam_for_sfn" {
-  name = "iam_for_sfn_%s"
+  name = %[1]q
 
   assume_role_policy = <<EOF
 {
@@ -213,23 +283,20 @@ resource "aws_iam_role" "iam_for_sfn" {
       "Principal": {
         "Service": "states.${data.aws_region.current.name}.amazonaws.com"
       },
-      "Action": "sts:AssumeRole"
+      "Action": "sts:AssumeRole",
+      "Sid": ""
     }
   ]
 }
 EOF
 }
-
-resource "aws_lambda_function" "lambda_function_test" {
-  filename      = "test-fixtures/lambdatest.zip"
-  function_name = "sfn-%s"
-  role          = "${aws_iam_role.iam_for_lambda.arn}"
-  handler       = "exports.example"
-  runtime       = "nodejs12.x"
+`, rName)
 }
 
-resource "aws_sfn_state_machine" "foo" {
-  name     = "test_sfn_%s"
+func testAccAWSSfnStateMachineConfig(rName string, rMaxAttempts int) string {
+	return testAccAWSSfnStateMachineConfigBase(rName) + fmt.Sprintf(`
+resource "aws_sfn_state_machine" "test" {
+  name     = "%s"
   role_arn = "${aws_iam_role.iam_for_sfn.arn}"
 
   definition = <<EOF
@@ -239,7 +306,7 @@ resource "aws_sfn_state_machine" "foo" {
   "States": {
     "HelloWorld": {
       "Type": "Task",
-      "Resource": "${aws_lambda_function.lambda_function_test.arn}",
+      "Resource": "${aws_lambda_function.test.arn}",
       "Retry": [
         {
           "ErrorEquals": ["States.ALL"],
@@ -254,100 +321,45 @@ resource "aws_sfn_state_machine" "foo" {
 }
 EOF
 }
-`, rName, rName, rName, rName, rName, rName, rMaxAttempts)
+`, rName, rMaxAttempts)
 }
 
-func testAccAWSSfnStateMachineConfigTags1(rName string, tag1Key, tag1Value string) string {
-	return fmt.Sprintf(`
-data "aws_partition" "current" {}
-
-data "aws_region" "current" {}
-
-resource "aws_iam_role_policy" "iam_policy_for_lambda" {
-  name = "iam_policy_for_lambda_%s"
-  role = "${aws_iam_role.iam_for_lambda.id}"
-  policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [{
-    "Effect": "Allow",
-    "Action": [
-      "logs:CreateLogGroup",
-      "logs:CreateLogStream",
-      "logs:PutLogEvents"
-    ],
-    "Resource": "arn:${data.aws_partition.current.partition}:logs:*:*:*"
-  }]
-}
-EOF
-}
-
-resource "aws_iam_role" "iam_for_lambda" {
-  name = "iam_for_lambda_%s"
-  assume_role_policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Action": "sts:AssumeRole",
-      "Principal": {
-        "Service": "lambda.amazonaws.com"
-      },
-      "Effect": "Allow",
-      "Sid": ""
-    }
-  ]
-}
-EOF
-}
+func testAccAWSSfnStateMachineConfigType(rName, typ string) string {
+	return testAccAWSSfnStateMachineConfigBase(rName) + fmt.Sprintf(`
+resource "aws_sfn_state_machine" "test" {
+  name     = %[1]q
+  type     = %[2]q
+  role_arn = "${aws_iam_role.iam_for_sfn.arn}"
 
-resource "aws_iam_role_policy" "iam_policy_for_sfn" {
-  name = "iam_policy_for_sfn_%s"
-  role = "${aws_iam_role.iam_for_sfn.id}"
-  policy = <<EOF
+  definition = <<EOF
 {
-  "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Effect": "Allow",
-      "Action": [
-        "lambda:InvokeFunction"
+  "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.0
+        }
       ],
-        "Resource": "*"
-      }
-  ]
-}
-EOF
-}
-
-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"
+      "End": true
     }
-  ]
+  }
 }
 EOF
 }
-
-resource "aws_lambda_function" "lambda_function_test" {
-  filename = "test-fixtures/lambdatest.zip"
-  function_name = "sfn-%s"
-  role = "${aws_iam_role.iam_for_lambda.arn}"
-  handler = "exports.example"
-  runtime = "nodejs12.x"
+`, rName, typ)
 }
 
-resource "aws_sfn_state_machine" "foo" {
-  name     = "test_sfn_%s"
+func testAccAWSSfnStateMachineConfigTags1(rName, tag1Key, tag1Value string) string {
+	return testAccAWSSfnStateMachineConfigBase(rName) + fmt.Sprintf(`
+resource "aws_sfn_state_machine" "test" {
+  name     = %[1]q
   role_arn = "${aws_iam_role.iam_for_sfn.arn}"
 
   definition = <<EOF
@@ -357,7 +369,7 @@ resource "aws_sfn_state_machine" "foo" {
   "States": {
     "HelloWorld": {
       "Type": "Task",
-      "Resource": "${aws_lambda_function.lambda_function_test.arn}",
+      "Resource": "${aws_lambda_function.test.arn}",
       "Retry": [
         {
           "ErrorEquals": ["States.ALL"],
@@ -372,103 +384,16 @@ resource "aws_sfn_state_machine" "foo" {
 }
 EOF
 tags = {
-	%q = %q
-}
-}
-`, rName, rName, rName, rName, rName, rName, tag1Key, tag1Value)
-}
-
-func testAccAWSSfnStateMachineConfigTags2(rName string, tag1Key, tag1Value, tag2Key, tag2Value string) string {
-	return fmt.Sprintf(`
-data "aws_partition" "current" {}
-
-data "aws_region" "current" {}
-
-resource "aws_iam_role_policy" "iam_policy_for_lambda" {
-  name = "iam_policy_for_lambda_%s"
-  role = "${aws_iam_role.iam_for_lambda.id}"
-  policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [{
-    "Effect": "Allow",
-    "Action": [
-      "logs:CreateLogGroup",
-      "logs:CreateLogStream",
-      "logs:PutLogEvents"
-    ],
-    "Resource": "arn:${data.aws_partition.current.partition}:logs:*:*:*"
-  }]
-}
-EOF
+	%[2]q = %[3]q
 }
-
-resource "aws_iam_role" "iam_for_lambda" {
-  name = "iam_for_lambda_%s"
-  assume_role_policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Action": "sts:AssumeRole",
-      "Principal": {
-        "Service": "lambda.amazonaws.com"
-      },
-      "Effect": "Allow",
-      "Sid": ""
-    }
-  ]
 }
-EOF
-}
-
-resource "aws_iam_role_policy" "iam_policy_for_sfn" {
-  name = "iam_policy_for_sfn_%s"
-  role = "${aws_iam_role.iam_for_sfn.id}"
-  policy = <<EOF
-{
-  "Version": "2012-10-17",
-  "Statement": [
-    {
-      "Effect": "Allow",
-      "Action": [
-        "lambda:InvokeFunction"
-      ],
-        "Resource": "*"
-      }
-  ]
-}
-EOF
-}
-
-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_lambda_function" "lambda_function_test" {
-  filename = "test-fixtures/lambdatest.zip"
-  function_name = "sfn-%s"
-  role = "${aws_iam_role.iam_for_lambda.arn}"
-  handler = "exports.example"
-  runtime = "nodejs12.x"
+`, rName, tag1Key, tag1Value)
 }
 
-resource "aws_sfn_state_machine" "foo" {
-  name     = "test_sfn_%s"
+func testAccAWSSfnStateMachineConfigTags2(rName, tag1Key, tag1Value, tag2Key, tag2Value string) string {
+	return testAccAWSSfnStateMachineConfigBase(rName) + fmt.Sprintf(`
+resource "aws_sfn_state_machine" "test" {
+  name     = %[1]q
   role_arn = "${aws_iam_role.iam_for_sfn.arn}"
 
   definition = <<EOF
@@ -478,7 +403,7 @@ resource "aws_sfn_state_machine" "foo" {
   "States": {
     "HelloWorld": {
       "Type": "Task",
-      "Resource": "${aws_lambda_function.lambda_function_test.arn}",
+      "Resource": "${aws_lambda_function.test.arn}",
       "Retry": [
         {
           "ErrorEquals": ["States.ALL"],
@@ -493,9 +418,9 @@ resource "aws_sfn_state_machine" "foo" {
 }
 EOF
 tags = {
-	%q = %q
-	%q = %q
+	%[2]q = %[3]q
+	%[4]q = %[5]q
 }
 }
-`, rName, rName, rName, rName, rName, rName, tag1Key, tag1Value, tag2Key, tag2Value)
+`, rName, tag1Key, tag1Value, tag2Key, tag2Value)
 }
diff --git a/website/docs/r/sfn_state_machine.html.markdown b/website/docs/r/sfn_state_machine.html.markdown
index 34163bef336..86ba2f58dc7 100644
--- a/website/docs/r/sfn_state_machine.html.markdown
+++ b/website/docs/r/sfn_state_machine.html.markdown
@@ -35,6 +35,32 @@ EOF
 }
 ```
 
+## Express State Machine Example
+
+```hcl
+# ...
+
+resource "aws_sfn_state_machine" "sfn_state_machine" {
+  name     = "my-state-machine"
+  role_arn = "${aws_iam_role.iam_for_sfn.arn}"
+  type     = "EXPRESS"
+
+  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.lambda.arn}",
+      "End": true
+    }
+  }
+}
+EOF
+}
+```
+
 ## Argument Reference
 
 The following arguments are supported:
@@ -42,6 +68,7 @@ The following arguments are supported:
 * `name` - (Required) The name of the state machine.
 * `definition` - (Required) The Amazon States Language definition of the state machine.
 * `role_arn` - (Required) The Amazon Resource Name (ARN) of the IAM role to use for this state machine.
+* `type` - (Optional) Determines whether a `Standard` or `Express` state machine is created. If not set, Standard is created.
 * `tags` - (Optional) Key-value mapping of resource tags
 
 ## Attributes Reference