diff --git a/.changelog/18561.txt b/.changelog/18561.txt
new file mode 100644
index 00000000000..077f159a0dc
--- /dev/null
+++ b/.changelog/18561.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_api_gateway_stage: Add `web_acl_arn` attribute
+```
\ No newline at end of file
diff --git a/internal/service/apigateway/deployment.go b/internal/service/apigateway/deployment.go
index d79ea6d790e..9efe1fbafcb 100644
--- a/internal/service/apigateway/deployment.go
+++ b/internal/service/apigateway/deployment.go
@@ -11,6 +11,8 @@ import (
 	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
+	"github.com/hashicorp/terraform-provider-aws/internal/flex"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
 func ResourceDeployment() *schema.Resource {
@@ -81,21 +83,15 @@ func resourceDeploymentCreate(d *schema.ResourceData, meta interface{}) error {
 	// Create the gateway
 	log.Printf("[DEBUG] Creating API Gateway Deployment")
 
-	variables := make(map[string]string)
-	for k, v := range d.Get("variables").(map[string]interface{}) {
-		variables[k] = v.(string)
-	}
-
-	var err error
 	deployment, err := conn.CreateDeployment(&apigateway.CreateDeploymentInput{
 		RestApiId:        aws.String(d.Get("rest_api_id").(string)),
 		StageName:        aws.String(d.Get("stage_name").(string)),
 		Description:      aws.String(d.Get("description").(string)),
 		StageDescription: aws.String(d.Get("stage_description").(string)),
-		Variables:        aws.StringMap(variables),
+		Variables:        flex.ExpandStringMap(d.Get("variables").(map[string]interface{})),
 	})
 	if err != nil {
-		return fmt.Errorf("Error creating API Gateway Deployment: %s", err)
+		return fmt.Errorf("Error creating API Gateway Deployment: %w", err)
 	}
 
 	d.SetId(aws.StringValue(deployment.Id))
@@ -188,14 +184,12 @@ func resourceDeploymentDelete(d *schema.ResourceData, meta interface{}) error {
 	// InvalidParameter: 1 validation error(s) found.
 	//  - minimum field size of 1, GetStageInput.StageName.
 	stageName := d.Get("stage_name").(string)
+	restApiId := d.Get("rest_api_id").(string)
 	if stageName != "" {
-		stage, err := conn.GetStage(&apigateway.GetStageInput{
-			StageName: aws.String(stageName),
-			RestApiId: aws.String(d.Get("rest_api_id").(string)),
-		})
+		stage, err := FindStageByName(conn, restApiId, stageName)
 
-		if err != nil && !tfawserr.ErrMessageContains(err, apigateway.ErrCodeNotFoundException, "") {
-			return fmt.Errorf("error getting referenced stage: %s", err)
+		if err != nil && !tfresource.NotFound(err) {
+			return fmt.Errorf("error getting referenced stage: %w", err)
 		}
 
 		if stage != nil && aws.StringValue(stage.DeploymentId) == d.Id() {
@@ -205,8 +199,8 @@ func resourceDeploymentDelete(d *schema.ResourceData, meta interface{}) error {
 
 	if shouldDeleteStage {
 		if _, err := conn.DeleteStage(&apigateway.DeleteStageInput{
-			StageName: aws.String(d.Get("stage_name").(string)),
-			RestApiId: aws.String(d.Get("rest_api_id").(string)),
+			StageName: aws.String(stageName),
+			RestApiId: aws.String(restApiId),
 		}); err == nil {
 			return nil
 		}
@@ -214,7 +208,7 @@ func resourceDeploymentDelete(d *schema.ResourceData, meta interface{}) error {
 
 	_, err := conn.DeleteDeployment(&apigateway.DeleteDeploymentInput{
 		DeploymentId: aws.String(d.Id()),
-		RestApiId:    aws.String(d.Get("rest_api_id").(string)),
+		RestApiId:    aws.String(restApiId),
 	})
 
 	if tfawserr.ErrMessageContains(err, apigateway.ErrCodeNotFoundException, "") {
@@ -222,7 +216,7 @@ func resourceDeploymentDelete(d *schema.ResourceData, meta interface{}) error {
 	}
 
 	if err != nil {
-		return fmt.Errorf("error deleting API Gateway Deployment (%s): %s", d.Id(), err)
+		return fmt.Errorf("error deleting API Gateway Deployment (%s): %w", d.Id(), err)
 	}
 
 	return nil
diff --git a/internal/service/apigateway/deployment_test.go b/internal/service/apigateway/deployment_test.go
index 3eb7e629cef..8e82b5115b2 100644
--- a/internal/service/apigateway/deployment_test.go
+++ b/internal/service/apigateway/deployment_test.go
@@ -87,7 +87,7 @@ func TestAccAPIGatewayDeployment_triggers(t *testing.T) {
 				Config: testAccDeploymentTriggersConfig("description1", "https://example.com"),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeploymentExists(resourceName, &deployment1),
-					testAccCheckDeploymentStageExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "description", "description1"),
 					resource.TestCheckResourceAttr(resourceName, "stage_description", "description1"),
 				),
@@ -100,7 +100,7 @@ func TestAccAPIGatewayDeployment_triggers(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeploymentExists(resourceName, &deployment2),
 					testAccCheckDeploymentRecreated(&deployment1, &deployment2),
-					testAccCheckDeploymentStageExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "description", "description1"),
 					resource.TestCheckResourceAttr(resourceName, "stage_description", "description1"),
 				),
@@ -110,7 +110,7 @@ func TestAccAPIGatewayDeployment_triggers(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeploymentExists(resourceName, &deployment3),
 					testAccCheckDeploymentNotRecreated(&deployment2, &deployment3),
-					testAccCheckDeploymentStageExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "description", "description1"),
 					resource.TestCheckResourceAttr(resourceName, "stage_description", "description1"),
 				),
@@ -120,7 +120,7 @@ func TestAccAPIGatewayDeployment_triggers(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeploymentExists(resourceName, &deployment4),
 					testAccCheckDeploymentRecreated(&deployment3, &deployment4),
-					testAccCheckDeploymentStageExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "description", "description2"),
 					resource.TestCheckResourceAttr(resourceName, "stage_description", "description2"),
 				),
@@ -172,7 +172,7 @@ func TestAccAPIGatewayDeployment_stageDescription(t *testing.T) {
 				Config: testAccDeploymentStageDescriptionConfig("description1"),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeploymentExists(resourceName, &deployment),
-					testAccCheckDeploymentStageExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "stage_description", "description1"),
 				),
 			},
@@ -195,7 +195,7 @@ func TestAccAPIGatewayDeployment_stageName(t *testing.T) {
 				Config: testAccDeploymentStageNameConfig("test"),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckDeploymentExists(resourceName, &deployment),
-					testAccCheckDeploymentStageExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "stage_name", "test"),
 				),
 			},
@@ -284,30 +284,6 @@ func testAccCheckDeploymentExists(n string, res *apigateway.Deployment) resource
 	}
 }
 
-func testAccCheckDeploymentStageExists(resourceName string, res *apigateway.Stage) resource.TestCheckFunc {
-	return func(s *terraform.State) error {
-		conn := acctest.Provider.Meta().(*conns.AWSClient).APIGatewayConn
-
-		rs, ok := s.RootModule().Resources[resourceName]
-		if !ok {
-			return fmt.Errorf("Deployment not found: %s", resourceName)
-		}
-
-		req := &apigateway.GetStageInput{
-			StageName: aws.String(rs.Primary.Attributes["stage_name"]),
-			RestApiId: aws.String(rs.Primary.Attributes["rest_api_id"]),
-		}
-		stage, err := conn.GetStage(req)
-		if err != nil {
-			return err
-		}
-
-		*res = *stage
-
-		return nil
-	}
-}
-
 func testAccCheckDeploymentDestroy(s *terraform.State) error {
 	conn := acctest.Provider.Meta().(*conns.AWSClient).APIGatewayConn
 
diff --git a/internal/service/apigateway/find.go b/internal/service/apigateway/find.go
new file mode 100644
index 00000000000..7ec0ee2bdff
--- /dev/null
+++ b/internal/service/apigateway/find.go
@@ -0,0 +1,34 @@
+package apigateway
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/apigateway"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
+)
+
+func FindStageByName(conn *apigateway.APIGateway, restApiId, name string) (*apigateway.Stage, error) {
+	input := &apigateway.GetStageInput{
+		RestApiId: aws.String(restApiId),
+		StageName: aws.String(name),
+	}
+
+	output, err := conn.GetStage(input)
+	if tfawserr.ErrCodeEquals(err, apigateway.ErrCodeNotFoundException) {
+		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
+}
diff --git a/internal/service/apigateway/method_settings.go b/internal/service/apigateway/method_settings.go
index 14a4f802f4a..e69e8bf648b 100644
--- a/internal/service/apigateway/method_settings.go
+++ b/internal/service/apigateway/method_settings.go
@@ -11,6 +11,7 @@ import (
 	"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/tfresource"
 )
 
 func ResourceMethodSettings() *schema.Resource {
@@ -97,14 +98,10 @@ func ResourceMethodSettings() *schema.Resource {
 							Computed: true,
 						},
 						"unauthorized_cache_control_header_strategy": {
-							Type:     schema.TypeString,
-							Optional: true,
-							ValidateFunc: validation.StringInSlice([]string{
-								apigateway.UnauthorizedCacheControlHeaderStrategyFailWith403,
-								apigateway.UnauthorizedCacheControlHeaderStrategySucceedWithResponseHeader,
-								apigateway.UnauthorizedCacheControlHeaderStrategySucceedWithoutResponseHeader,
-							}, false),
-							Computed: true,
+							Type:         schema.TypeString,
+							Optional:     true,
+							ValidateFunc: validation.StringInSlice(apigateway.UnauthorizedCacheControlHeaderStrategy_Values(), false),
+							Computed:     true,
 						},
 					},
 				},
@@ -137,14 +134,9 @@ func flattenMethodSettings(settings *apigateway.MethodSetting) []interface{} {
 func resourceMethodSettingsRead(d *schema.ResourceData, meta interface{}) error {
 	conn := meta.(*conns.AWSClient).APIGatewayConn
 
-	input := &apigateway.GetStageInput{
-		RestApiId: aws.String(d.Get("rest_api_id").(string)),
-		StageName: aws.String(d.Get("stage_name").(string)),
-	}
-
-	stage, err := conn.GetStage(input)
+	stage, err := FindStageByName(conn, d.Get("rest_api_id").(string), d.Get("stage_name").(string))
 
-	if !d.IsNewResource() && tfawserr.ErrCodeEquals(err, apigateway.ErrCodeNotFoundException) {
+	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] API Gateway Stage Method Settings (%s) not found, removing from state", d.Id())
 		d.SetId("")
 		return nil
diff --git a/internal/service/apigateway/method_settings_test.go b/internal/service/apigateway/method_settings_test.go
index 9583b10fce4..6eafa4140e0 100644
--- a/internal/service/apigateway/method_settings_test.go
+++ b/internal/service/apigateway/method_settings_test.go
@@ -5,7 +5,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/apigateway"
 	sdkacctest "github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
@@ -13,6 +12,7 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/acctest"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
 	tfapigateway "github.com/hashicorp/terraform-provider-aws/internal/service/apigateway"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
 func TestAccAPIGatewayMethodSettings_basic(t *testing.T) {
@@ -29,7 +29,7 @@ func TestAccAPIGatewayMethodSettings_basic(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsLoggingLevelConfig(rName, "INFO"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.logging_level", "INFO"),
 				),
@@ -58,7 +58,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cacheDataEncrypted(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCacheDataEncryptedConfig(rName, true),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.cache_data_encrypted", "true"),
 				),
@@ -66,7 +66,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cacheDataEncrypted(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCacheDataEncryptedConfig(rName, false),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.cache_data_encrypted", "false"),
 				),
@@ -95,7 +95,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cacheTTLInSeconds(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCacheTTLInSecondsConfig(rName, 0),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.cache_ttl_in_seconds", "0"),
 				),
@@ -103,7 +103,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cacheTTLInSeconds(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCacheTTLInSecondsConfig(rName, 1),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.cache_ttl_in_seconds", "1"),
 				),
@@ -111,7 +111,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cacheTTLInSeconds(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCacheTTLInSecondsConfig(rName, 2),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.cache_ttl_in_seconds", "2"),
 				),
@@ -140,7 +140,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cachingEnabled(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCachingEnabledConfig(rName, true),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.caching_enabled", "true"),
 				),
@@ -148,7 +148,7 @@ func TestAccAPIGatewayMethodSettings_Settings_cachingEnabled(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsCachingEnabledConfig(rName, false),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.caching_enabled", "false"),
 				),
@@ -177,7 +177,7 @@ func TestAccAPIGatewayMethodSettings_Settings_dataTraceEnabled(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsDataTraceEnabledConfig(rName, true),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.data_trace_enabled", "true"),
 				),
@@ -185,7 +185,7 @@ func TestAccAPIGatewayMethodSettings_Settings_dataTraceEnabled(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsDataTraceEnabledConfig(rName, false),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.data_trace_enabled", "false"),
 				),
@@ -214,7 +214,7 @@ func TestAccAPIGatewayMethodSettings_Settings_loggingLevel(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsLoggingLevelConfig(rName, "INFO"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					testAccCheckMethodSettings_loggingLevel(&stage1, "test/GET", "INFO"),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.logging_level", "INFO"),
@@ -223,7 +223,7 @@ func TestAccAPIGatewayMethodSettings_Settings_loggingLevel(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsLoggingLevelConfig(rName, "OFF"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					testAccCheckMethodSettings_loggingLevel(&stage2, "test/GET", "OFF"),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.logging_level", "OFF"),
@@ -253,7 +253,7 @@ func TestAccAPIGatewayMethodSettings_Settings_metricsEnabled(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsMetricsEnabledConfig(rName, true),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					testAccCheckMethodSettings_metricsEnabled(&stage1, "test/GET", true),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.metrics_enabled", "true"),
@@ -262,7 +262,7 @@ func TestAccAPIGatewayMethodSettings_Settings_metricsEnabled(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsMetricsEnabledConfig(rName, false),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					testAccCheckMethodSettings_metricsEnabled(&stage2, "test/GET", false),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.metrics_enabled", "false"),
@@ -292,7 +292,7 @@ func TestAccAPIGatewayMethodSettings_Settings_multiple(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsMultipleConfig(rName, "INFO", true),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					testAccCheckMethodSettings_metricsEnabled(&stage1, "test/GET", true),
 					testAccCheckMethodSettings_loggingLevel(&stage1, "test/GET", "INFO"),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
@@ -303,7 +303,7 @@ func TestAccAPIGatewayMethodSettings_Settings_multiple(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsMultipleConfig(rName, "OFF", false),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					testAccCheckMethodSettings_metricsEnabled(&stage2, "test/GET", false),
 					testAccCheckMethodSettings_loggingLevel(&stage2, "test/GET", "OFF"),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
@@ -335,7 +335,7 @@ func TestAccAPIGatewayMethodSettings_Settings_requireAuthorizationForCacheContro
 			{
 				Config: testAccMethodSettingsSettingsRequireAuthorizationForCacheControlConfig(rName, true),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.require_authorization_for_cache_control", "true"),
 				),
@@ -343,7 +343,7 @@ func TestAccAPIGatewayMethodSettings_Settings_requireAuthorizationForCacheContro
 			{
 				Config: testAccMethodSettingsSettingsRequireAuthorizationForCacheControlConfig(rName, false),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.require_authorization_for_cache_control", "false"),
 				),
@@ -372,7 +372,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingBurstLimit(t *testing.T)
 			{
 				Config: testAccMethodSettingsSettingsThrottlingBurstLimitConfig(rName, 1),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_burst_limit", "1"),
 				),
@@ -380,7 +380,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingBurstLimit(t *testing.T)
 			{
 				Config: testAccMethodSettingsSettingsThrottlingBurstLimitConfig(rName, 2),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_burst_limit", "2"),
 				),
@@ -410,7 +410,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingBurstLimitDisabledByDefa
 			{
 				Config: testAccMethodSettingsSettingsLoggingLevelConfig(rName, "INFO"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_burst_limit", "-1"),
 				),
@@ -424,7 +424,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingBurstLimitDisabledByDefa
 			{
 				Config: testAccMethodSettingsSettingsThrottlingBurstLimitConfig(rName, 1),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_burst_limit", "1"),
 				),
@@ -447,7 +447,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingRateLimit(t *testing.T)
 			{
 				Config: testAccMethodSettingsSettingsThrottlingRateLimitConfig(rName, 1.1),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_rate_limit", "1.1"),
 				),
@@ -455,7 +455,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingRateLimit(t *testing.T)
 			{
 				Config: testAccMethodSettingsSettingsThrottlingRateLimitConfig(rName, 2.2),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_rate_limit", "2.2"),
 				),
@@ -485,7 +485,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingRateLimitDisabledByDefau
 			{
 				Config: testAccMethodSettingsSettingsLoggingLevelConfig(rName, "INFO"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_rate_limit", "-1"),
 				),
@@ -499,7 +499,7 @@ func TestAccAPIGatewayMethodSettings_Settings_throttlingRateLimitDisabledByDefau
 			{
 				Config: testAccMethodSettingsSettingsThrottlingRateLimitConfig(rName, 1.1),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.throttling_rate_limit", "1.1"),
 				),
@@ -522,7 +522,7 @@ func TestAccAPIGatewayMethodSettings_Settings_unauthorizedCacheControlHeaderStra
 			{
 				Config: testAccMethodSettingsSettingsUnauthorizedCacheControlHeaderStrategyConfig(rName, "SUCCEED_WITH_RESPONSE_HEADER"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage1),
+					testAccCheckStageExists(resourceName, &stage1),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.unauthorized_cache_control_header_strategy", "SUCCEED_WITH_RESPONSE_HEADER"),
 				),
@@ -530,7 +530,7 @@ func TestAccAPIGatewayMethodSettings_Settings_unauthorizedCacheControlHeaderStra
 			{
 				Config: testAccMethodSettingsSettingsUnauthorizedCacheControlHeaderStrategyConfig(rName, "SUCCEED_WITHOUT_RESPONSE_HEADER"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage2),
+					testAccCheckStageExists(resourceName, &stage2),
 					resource.TestCheckResourceAttr(resourceName, "settings.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "settings.0.unauthorized_cache_control_header_strategy", "SUCCEED_WITHOUT_RESPONSE_HEADER"),
 				),
@@ -552,11 +552,11 @@ func testAccCheckMethodSettings_metricsEnabled(conf *apigateway.Stage, path stri
 			return fmt.Errorf("Expected to find method settings for %q", path)
 		}
 
-		if expected && *settings.MetricsEnabled != expected {
-			return fmt.Errorf("Expected metrics to be enabled, got %t", *settings.MetricsEnabled)
+		if expected && aws.BoolValue(settings.MetricsEnabled) != expected {
+			return fmt.Errorf("Expected metrics to be enabled, got %t", aws.BoolValue(settings.MetricsEnabled))
 		}
-		if !expected && *settings.MetricsEnabled != expected {
-			return fmt.Errorf("Expected metrics to be disabled, got %t", *settings.MetricsEnabled)
+		if !expected && aws.BoolValue(settings.MetricsEnabled) != expected {
+			return fmt.Errorf("Expected metrics to be disabled, got %t", aws.BoolValue(settings.MetricsEnabled))
 		}
 
 		return nil
@@ -570,8 +570,8 @@ func testAccCheckMethodSettings_loggingLevel(conf *apigateway.Stage, path string
 			return fmt.Errorf("Expected to find method settings for %q", path)
 		}
 
-		if *settings.LoggingLevel != expectedLevel {
-			return fmt.Errorf("Expected logging level to match %q, got %q", expectedLevel, *settings.LoggingLevel)
+		if aws.StringValue(settings.LoggingLevel) != expectedLevel {
+			return fmt.Errorf("Expected logging level to match %q, got %q", expectedLevel, aws.StringValue(settings.LoggingLevel))
 		}
 
 		return nil
@@ -592,7 +592,7 @@ func TestAccAPIGatewayMethodSettings_disappears(t *testing.T) {
 			{
 				Config: testAccMethodSettingsSettingsLoggingLevelConfig(rName, "INFO"),
 				Check: resource.ComposeTestCheckFunc(
-					testAccCheckMethodSettingsExists(resourceName, &stage),
+					testAccCheckStageExists(resourceName, &stage),
 					acctest.CheckResourceDisappears(acctest.Provider, tfapigateway.ResourceMethodSettings(), resourceName),
 				),
 				ExpectNonEmptyPlan: true,
@@ -601,34 +601,6 @@ func TestAccAPIGatewayMethodSettings_disappears(t *testing.T) {
 	})
 }
 
-func testAccCheckMethodSettingsExists(n string, res *apigateway.Stage) resource.TestCheckFunc {
-	return func(s *terraform.State) error {
-		rs, ok := s.RootModule().Resources[n]
-		if !ok {
-			return fmt.Errorf("Not found: %s", n)
-		}
-
-		if rs.Primary.ID == "" {
-			return fmt.Errorf("No API Gateway Stage ID is set")
-		}
-
-		conn := acctest.Provider.Meta().(*conns.AWSClient).APIGatewayConn
-
-		req := &apigateway.GetStageInput{
-			StageName: aws.String(rs.Primary.Attributes["stage_name"]),
-			RestApiId: aws.String(rs.Primary.Attributes["rest_api_id"]),
-		}
-		out, err := conn.GetStage(req)
-		if err != nil {
-			return err
-		}
-
-		*res = *out
-
-		return nil
-	}
-}
-
 func testAccCheckMethodSettingsDestroy(s *terraform.State) error {
 	conn := acctest.Provider.Meta().(*conns.AWSClient).APIGatewayConn
 
@@ -637,22 +609,16 @@ func testAccCheckMethodSettingsDestroy(s *terraform.State) error {
 			continue
 		}
 
-		req := &apigateway.GetStageInput{
-			StageName: aws.String(rs.Primary.Attributes["stage_name"]),
-			RestApiId: aws.String(rs.Primary.Attributes["rest_api_id"]),
-		}
-		out, err := conn.GetStage(req)
-		if err == nil {
-			return fmt.Errorf("API Gateway Stage still exists: %s", out)
+		_, err := tfapigateway.FindStageByName(conn, rs.Primary.Attributes["rest_api_id"], rs.Primary.Attributes["stage_name"])
+		if tfresource.NotFound(err) {
+			continue
 		}
 
-		awsErr, ok := err.(awserr.Error)
-		if !ok {
-			return err
-		}
-		if awsErr.Code() != "NotFoundException" {
+		if err != nil {
 			return err
 		}
+
+		return fmt.Errorf("API Gateway Stage %s still exists", rs.Primary.ID)
 	}
 
 	return nil
diff --git a/internal/service/apigateway/stage.go b/internal/service/apigateway/stage.go
index ed66f9835b4..2af715f3bb3 100644
--- a/internal/service/apigateway/stage.go
+++ b/internal/service/apigateway/stage.go
@@ -4,17 +4,17 @@ import (
 	"fmt"
 	"log"
 	"strings"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/arn"
 	"github.com/aws/aws-sdk-go/service/apigateway"
 	"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"
+	"github.com/hashicorp/terraform-provider-aws/internal/flex"
 	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"
 )
 
@@ -47,8 +47,9 @@ func ResourceStage() *schema.Resource {
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"destination_arn": {
-							Type:     schema.TypeString,
-							Required: true,
+							Type:         schema.TypeString,
+							Required:     true,
+							ValidateFunc: verify.ValidARN,
 						},
 						"format": {
 							Type:     schema.TypeString,
@@ -62,18 +63,9 @@ func ResourceStage() *schema.Resource {
 				Optional: true,
 			},
 			"cache_cluster_size": {
-				Type:     schema.TypeString,
-				Optional: true,
-				ValidateFunc: validation.StringInSlice([]string{
-					apigateway.CacheClusterSize05,
-					apigateway.CacheClusterSize16,
-					apigateway.CacheClusterSize61,
-					apigateway.CacheClusterSize118,
-					apigateway.CacheClusterSize135,
-					apigateway.CacheClusterSize237,
-					apigateway.CacheClusterSize284,
-					apigateway.CacheClusterSize582,
-				}, true),
+				Type:         schema.TypeString,
+				Optional:     true,
+				ValidateFunc: validation.StringInSlice(apigateway.CacheClusterSize_Values(), true),
 			},
 			"client_certificate_id": {
 				Type:     schema.TypeString,
@@ -124,6 +116,10 @@ func ResourceStage() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
+			"web_acl_arn": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
 		},
 
 		CustomizeDiff: verify.SetTagsDiff,
@@ -135,9 +131,11 @@ func resourceStageCreate(d *schema.ResourceData, meta interface{}) error {
 	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
 	tags := defaultTagsConfig.MergeTags(tftags.New(d.Get("tags").(map[string]interface{})))
 
+	respApiId := d.Get("rest_api_id").(string)
+	stageName := d.Get("stage_name").(string)
 	input := apigateway.CreateStageInput{
-		RestApiId:    aws.String(d.Get("rest_api_id").(string)),
-		StageName:    aws.String(d.Get("stage_name").(string)),
+		RestApiId:    aws.String(respApiId),
+		StageName:    aws.String(stageName),
 		DeploymentId: aws.String(d.Get("deployment_id").(string)),
 	}
 
@@ -160,49 +158,33 @@ func resourceStageCreate(d *schema.ResourceData, meta interface{}) error {
 		input.DocumentationVersion = aws.String(v.(string))
 	}
 	if vars, ok := d.GetOk("variables"); ok {
-		variables := make(map[string]string)
-		for k, v := range vars.(map[string]interface{}) {
-			variables[k] = v.(string)
-		}
-		input.Variables = aws.StringMap(variables)
+		input.Variables = flex.ExpandStringMap(vars.(map[string]interface{}))
 	}
 	if len(tags) > 0 {
 		input.Tags = Tags(tags.IgnoreAWS())
 	}
 
-	out, err := conn.CreateStage(&input)
+	_, err := conn.CreateStage(&input)
 	if err != nil {
 		return fmt.Errorf("Error creating API Gateway Stage: %s", err)
 	}
 
-	d.SetId(fmt.Sprintf("ags-%s-%s", d.Get("rest_api_id").(string), d.Get("stage_name").(string)))
-
-	if waitForCache && out != nil && aws.StringValue(out.CacheClusterStatus) != apigateway.CacheClusterStatusNotAvailable {
-		stateConf := &resource.StateChangeConf{
-			Pending: []string{
-				apigateway.CacheClusterStatusCreateInProgress,
-				apigateway.CacheClusterStatusDeleteInProgress,
-				apigateway.CacheClusterStatusFlushInProgress,
-			},
-			Target: []string{apigateway.CacheClusterStatusAvailable},
-			Refresh: apiGatewayStageCacheRefreshFunc(conn,
-				d.Get("rest_api_id").(string),
-				d.Get("stage_name").(string)),
-			Timeout: 90 * time.Minute,
-		}
+	d.SetId(fmt.Sprintf("ags-%s-%s", respApiId, stageName))
 
-		_, err := stateConf.WaitForState()
+	if waitForCache {
+		_, err := waitStageCacheAvailable(conn, respApiId, stageName)
 		if err != nil {
-			return err
+			return fmt.Errorf("error waiting for API Gateway Stage (%s) to be available: %w", d.Id(), err)
 		}
 	}
 
-	if _, ok := d.GetOk("client_certificate_id"); ok {
-		return resourceStageUpdate(d, meta)
-	}
-	if _, ok := d.GetOk("access_log_settings"); ok {
+	_, certOk := d.GetOk("client_certificate_id")
+	_, logsOk := d.GetOk("access_log_settings")
+
+	if certOk || logsOk {
 		return resourceStageUpdate(d, meta)
 	}
+
 	return resourceStageRead(d, meta)
 }
 
@@ -214,13 +196,9 @@ func resourceStageRead(d *schema.ResourceData, meta interface{}) error {
 	log.Printf("[DEBUG] Reading API Gateway Stage %s", d.Id())
 	restApiId := d.Get("rest_api_id").(string)
 	stageName := d.Get("stage_name").(string)
-	input := apigateway.GetStageInput{
-		RestApiId: aws.String(restApiId),
-		StageName: aws.String(stageName),
-	}
-	stage, err := conn.GetStage(&input)
+	stage, err := FindStageByName(conn, restApiId, stageName)
 
-	if !d.IsNewResource() && tfawserr.ErrCodeEquals(err, apigateway.ErrCodeNotFoundException) {
+	if !d.IsNewResource() && tfresource.NotFound(err) {
 		log.Printf("[WARN] API Gateway Stage (%s) not found, removing from state", d.Id())
 		d.SetId("")
 		return nil
@@ -250,6 +228,7 @@ func resourceStageRead(d *schema.ResourceData, meta interface{}) error {
 	d.Set("description", stage.Description)
 	d.Set("documentation_version", stage.DocumentationVersion)
 	d.Set("xray_tracing_enabled", stage.TracingEnabled)
+	d.Set("web_acl_arn", stage.WebAclArn)
 
 	tags := KeyValueTags(stage.Tags).IgnoreAWS().IgnoreConfig(ignoreTagsConfig)
 
@@ -291,131 +270,119 @@ func resourceStageRead(d *schema.ResourceData, meta interface{}) error {
 func resourceStageUpdate(d *schema.ResourceData, meta interface{}) error {
 	conn := meta.(*conns.AWSClient).APIGatewayConn
 
+	respApiId := d.Get("rest_api_id").(string)
+	stageName := d.Get("stage_name").(string)
+
 	stageArn := arn.ARN{
 		Partition: meta.(*conns.AWSClient).Partition,
 		Region:    meta.(*conns.AWSClient).Region,
 		Service:   "apigateway",
-		Resource:  fmt.Sprintf("/restapis/%s/stages/%s", d.Get("rest_api_id").(string), d.Get("stage_name").(string)),
+		Resource:  fmt.Sprintf("/restapis/%s/stages/%s", respApiId, stageName),
 	}.String()
 	if d.HasChange("tags_all") {
 		o, n := d.GetChange("tags_all")
 		if err := UpdateTags(conn, stageArn, o, n); err != nil {
-			return fmt.Errorf("error updating tags: %s", err)
+			return fmt.Errorf("error updating tags: %w", err)
 		}
 	}
 
-	operations := make([]*apigateway.PatchOperation, 0)
-	waitForCache := false
-	if d.HasChange("cache_cluster_enabled") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/cacheClusterEnabled"),
-			Value: aws.String(fmt.Sprintf("%t", d.Get("cache_cluster_enabled").(bool))),
-		})
-		waitForCache = true
-	}
-	if d.HasChange("cache_cluster_size") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/cacheClusterSize"),
-			Value: aws.String(d.Get("cache_cluster_size").(string)),
-		})
-		waitForCache = true
-	}
-	if d.HasChange("client_certificate_id") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/clientCertificateId"),
-			Value: aws.String(d.Get("client_certificate_id").(string)),
-		})
-	}
-	if d.HasChange("deployment_id") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/deploymentId"),
-			Value: aws.String(d.Get("deployment_id").(string)),
-		})
-	}
-	if d.HasChange("description") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/description"),
-			Value: aws.String(d.Get("description").(string)),
-		})
-	}
-	if d.HasChange("xray_tracing_enabled") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/tracingEnabled"),
-			Value: aws.String(fmt.Sprintf("%t", d.Get("xray_tracing_enabled").(bool))),
-		})
-	}
-	if d.HasChange("documentation_version") {
-		operations = append(operations, &apigateway.PatchOperation{
-			Op:    aws.String(apigateway.OpReplace),
-			Path:  aws.String("/documentationVersion"),
-			Value: aws.String(d.Get("documentation_version").(string)),
-		})
-	}
-	if d.HasChange("variables") {
-		o, n := d.GetChange("variables")
-		oldV := o.(map[string]interface{})
-		newV := n.(map[string]interface{})
-		operations = append(operations, diffVariablesOps(oldV, newV)...)
-	}
-	if d.HasChange("access_log_settings") {
-		accessLogSettings := d.Get("access_log_settings").([]interface{})
-		if len(accessLogSettings) == 1 {
-			operations = append(operations,
-				&apigateway.PatchOperation{
-					Op:    aws.String(apigateway.OpReplace),
-					Path:  aws.String("/accessLogSettings/destinationArn"),
-					Value: aws.String(d.Get("access_log_settings.0.destination_arn").(string)),
-				}, &apigateway.PatchOperation{
-					Op:    aws.String(apigateway.OpReplace),
-					Path:  aws.String("/accessLogSettings/format"),
-					Value: aws.String(d.Get("access_log_settings.0.format").(string)),
-				})
-		} else if len(accessLogSettings) == 0 {
+	if d.HasChangesExcept("tags", "tags_all") {
+		operations := make([]*apigateway.PatchOperation, 0)
+		waitForCache := false
+		if d.HasChange("cache_cluster_enabled") {
 			operations = append(operations, &apigateway.PatchOperation{
-				Op:   aws.String(apigateway.OpRemove),
-				Path: aws.String("/accessLogSettings"),
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/cacheClusterEnabled"),
+				Value: aws.String(fmt.Sprintf("%t", d.Get("cache_cluster_enabled").(bool))),
 			})
+			waitForCache = true
 		}
-	}
-
-	input := apigateway.UpdateStageInput{
-		RestApiId:       aws.String(d.Get("rest_api_id").(string)),
-		StageName:       aws.String(d.Get("stage_name").(string)),
-		PatchOperations: operations,
-	}
-	log.Printf("[DEBUG] Updating API Gateway Stage: %s", input)
-	out, err := conn.UpdateStage(&input)
-	if err != nil {
-		return fmt.Errorf("Updating API Gateway Stage failed: %s", err)
-	}
-
-	if waitForCache && out != nil && aws.StringValue(out.CacheClusterStatus) != apigateway.CacheClusterStatusNotAvailable {
-		stateConf := &resource.StateChangeConf{
-			Pending: []string{
-				apigateway.CacheClusterStatusCreateInProgress,
-				apigateway.CacheClusterStatusFlushInProgress,
-			},
-			Target: []string{
-				apigateway.CacheClusterStatusAvailable,
-				// There's an AWS API bug (raised & confirmed in Sep 2016 by support)
-				// which causes the stage to remain in deletion state forever
-				apigateway.CacheClusterStatusDeleteInProgress,
-			},
-			Refresh: apiGatewayStageCacheRefreshFunc(conn,
-				d.Get("rest_api_id").(string),
-				d.Get("stage_name").(string)),
-			Timeout: 30 * time.Minute,
+		if d.HasChange("cache_cluster_size") {
+			operations = append(operations, &apigateway.PatchOperation{
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/cacheClusterSize"),
+				Value: aws.String(d.Get("cache_cluster_size").(string)),
+			})
+			waitForCache = true
+		}
+		if d.HasChange("client_certificate_id") {
+			operations = append(operations, &apigateway.PatchOperation{
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/clientCertificateId"),
+				Value: aws.String(d.Get("client_certificate_id").(string)),
+			})
+		}
+		if d.HasChange("deployment_id") {
+			operations = append(operations, &apigateway.PatchOperation{
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/deploymentId"),
+				Value: aws.String(d.Get("deployment_id").(string)),
+			})
+		}
+		if d.HasChange("description") {
+			operations = append(operations, &apigateway.PatchOperation{
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/description"),
+				Value: aws.String(d.Get("description").(string)),
+			})
+		}
+		if d.HasChange("xray_tracing_enabled") {
+			operations = append(operations, &apigateway.PatchOperation{
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/tracingEnabled"),
+				Value: aws.String(fmt.Sprintf("%t", d.Get("xray_tracing_enabled").(bool))),
+			})
+		}
+		if d.HasChange("documentation_version") {
+			operations = append(operations, &apigateway.PatchOperation{
+				Op:    aws.String(apigateway.OpReplace),
+				Path:  aws.String("/documentationVersion"),
+				Value: aws.String(d.Get("documentation_version").(string)),
+			})
+		}
+		if d.HasChange("variables") {
+			o, n := d.GetChange("variables")
+			oldV := o.(map[string]interface{})
+			newV := n.(map[string]interface{})
+			operations = append(operations, diffVariablesOps(oldV, newV)...)
+		}
+		if d.HasChange("access_log_settings") {
+			accessLogSettings := d.Get("access_log_settings").([]interface{})
+			if len(accessLogSettings) == 1 {
+				operations = append(operations,
+					&apigateway.PatchOperation{
+						Op:    aws.String(apigateway.OpReplace),
+						Path:  aws.String("/accessLogSettings/destinationArn"),
+						Value: aws.String(d.Get("access_log_settings.0.destination_arn").(string)),
+					}, &apigateway.PatchOperation{
+						Op:    aws.String(apigateway.OpReplace),
+						Path:  aws.String("/accessLogSettings/format"),
+						Value: aws.String(d.Get("access_log_settings.0.format").(string)),
+					})
+			} else if len(accessLogSettings) == 0 {
+				operations = append(operations, &apigateway.PatchOperation{
+					Op:   aws.String(apigateway.OpRemove),
+					Path: aws.String("/accessLogSettings"),
+				})
+			}
 		}
 
-		_, err := stateConf.WaitForState()
+		input := apigateway.UpdateStageInput{
+			RestApiId:       aws.String(respApiId),
+			StageName:       aws.String(stageName),
+			PatchOperations: operations,
+		}
+		log.Printf("[DEBUG] Updating API Gateway Stage: %s", input)
+		_, err := conn.UpdateStage(&input)
 		if err != nil {
-			return err
+			return fmt.Errorf("Updating API Gateway Stage failed: %w", err)
+		}
+
+		if waitForCache {
+			_, err := waitStageCacheUpdated(conn, respApiId, stageName)
+			if err != nil {
+				return fmt.Errorf("error waiting for API Gateway Stage (%s) to be updated: %w", d.Id(), err)
+			}
 		}
 	}
 
@@ -454,21 +421,6 @@ func diffVariablesOps(oldVars, newVars map[string]interface{}) []*apigateway.Pat
 	return ops
 }
 
-func apiGatewayStageCacheRefreshFunc(conn *apigateway.APIGateway, apiId, stageName string) func() (interface{}, string, error) {
-	return func() (interface{}, string, error) {
-		input := apigateway.GetStageInput{
-			RestApiId: aws.String(apiId),
-			StageName: aws.String(stageName),
-		}
-		out, err := conn.GetStage(&input)
-		if err != nil {
-			return 42, "", err
-		}
-
-		return out, *out.CacheClusterStatus, nil
-	}
-}
-
 func resourceStageDelete(d *schema.ResourceData, meta interface{}) error {
 	conn := meta.(*conns.AWSClient).APIGatewayConn
 	log.Printf("[DEBUG] Deleting API Gateway Stage: %s", d.Id())
diff --git a/internal/service/apigateway/stage_test.go b/internal/service/apigateway/stage_test.go
index 9f5280af891..dcc910ef2e2 100644
--- a/internal/service/apigateway/stage_test.go
+++ b/internal/service/apigateway/stage_test.go
@@ -5,8 +5,6 @@ import (
 	"regexp"
 	"testing"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/service/apigateway"
 	sdkacctest "github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
@@ -14,6 +12,7 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/acctest"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
 	tfapigateway "github.com/hashicorp/terraform-provider-aws/internal/service/apigateway"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
 func TestAccAPIGatewayStage_basic(t *testing.T) {
@@ -28,18 +27,17 @@ func TestAccAPIGatewayStage_basic(t *testing.T) {
 		CheckDestroy: testAccCheckStageDestroy,
 		Steps: []resource.TestStep{
 			{
-				Config: testAccStageConfig_basic(rName),
+				Config: testAccStageConfigBasic(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
 					resource.TestCheckResourceAttr(resourceName, "stage_name", "prod"),
-					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "true"),
-					resource.TestCheckResourceAttr(resourceName, "cache_cluster_size", "0.5"),
-					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
-					resource.TestCheckResourceAttr(resourceName, "tags.Name", "tf-test"),
 					resource.TestCheckResourceAttrSet(resourceName, "execution_arn"),
 					resource.TestCheckResourceAttrSet(resourceName, "invoke_url"),
-					resource.TestCheckResourceAttr(resourceName, "xray_tracing_enabled", "true"),
+					resource.TestCheckResourceAttr(resourceName, "description", ""),
+					resource.TestCheckResourceAttr(resourceName, "variables.%", "0"),
+					resource.TestCheckResourceAttr(resourceName, "xray_tracing_enabled", "false"),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "0"),
 				),
 			},
 			{
@@ -49,32 +47,83 @@ func TestAccAPIGatewayStage_basic(t *testing.T) {
 				ImportStateVerify: true,
 			},
 			{
-				Config: testAccStageConfig_updated(rName),
+				Config: testAccStageConfigUpdated(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
 					resource.TestCheckResourceAttr(resourceName, "stage_name", "prod"),
-					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "false"),
-					resource.TestCheckResourceAttr(resourceName, "tags.Name", "tf-test"),
-					resource.TestCheckResourceAttr(resourceName, "tags.%", "2"),
-					resource.TestCheckResourceAttr(resourceName, "tags.Name", "tf-test"),
-					resource.TestCheckResourceAttr(resourceName, "tags.ExtraName", "tf-test"),
-					resource.TestCheckResourceAttr(resourceName, "xray_tracing_enabled", "false"),
+					resource.TestCheckResourceAttrSet(resourceName, "execution_arn"),
+					resource.TestCheckResourceAttrSet(resourceName, "invoke_url"),
+					resource.TestCheckResourceAttr(resourceName, "description", "Hello world"),
+					resource.TestCheckResourceAttr(resourceName, "variables.%", "2"),
+					resource.TestCheckResourceAttr(resourceName, "variables.one", "1"),
+					resource.TestCheckResourceAttr(resourceName, "variables.three", "3"),
+					resource.TestCheckResourceAttr(resourceName, "xray_tracing_enabled", "true"),
 				),
 			},
 			{
-				Config: testAccStageConfig_basic(rName),
+				Config: testAccStageConfigBasic(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
 					resource.TestCheckResourceAttr(resourceName, "stage_name", "prod"),
-					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "true"),
-					resource.TestCheckResourceAttr(resourceName, "cache_cluster_size", "0.5"),
-					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
-					resource.TestCheckResourceAttr(resourceName, "tags.Name", "tf-test"),
 					resource.TestCheckResourceAttrSet(resourceName, "execution_arn"),
 					resource.TestCheckResourceAttrSet(resourceName, "invoke_url"),
-					resource.TestCheckResourceAttr(resourceName, "xray_tracing_enabled", "true"),
+					resource.TestCheckResourceAttr(resourceName, "description", ""),
+					resource.TestCheckResourceAttr(resourceName, "variables.%", "0"),
+					resource.TestCheckResourceAttr(resourceName, "xray_tracing_enabled", "false"),
+				),
+			},
+		},
+	})
+}
+
+func TestAccAPIGatewayStage_cache(t *testing.T) {
+	var conf apigateway.Stage
+	rName := sdkacctest.RandString(5)
+	resourceName := "aws_api_gateway_stage.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t); acctest.PreCheckAPIGatewayTypeEDGE(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, apigateway.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckStageDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStageConfigBasic(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "false"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccStageImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccStageConfigCacheConfig(rName, "0.5"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "cache_cluster_size", "0.5"),
+					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "true"),
+				),
+			},
+
+			{
+				Config: testAccStageConfigCacheConfig(rName, "1.6"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "cache_cluster_size", "1.6"),
+					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "true"),
+				),
+			},
+			{
+				Config: testAccStageConfigBasic(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "cache_cluster_enabled", "false"),
 				),
 			},
 		},
@@ -109,6 +158,52 @@ func TestAccAPIGatewayStage_Disappears_referencingDeployment(t *testing.T) {
 	})
 }
 
+func TestAccAPIGatewayStage_tags(t *testing.T) {
+	var conf apigateway.Stage
+	rName := sdkacctest.RandString(5)
+	resourceName := "aws_api_gateway_stage.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t); acctest.PreCheckAPIGatewayTypeEDGE(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, apigateway.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckStageDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStageConfigTags1(rName, "key1", "value1"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key1", "value1"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccStageImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
+			{
+				Config: testAccStageConfigTags2(rName, "key1", "value1updated", "key2", "value2"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "2"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key1", "value1updated"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key2", "value2"),
+				),
+			},
+			{
+				Config: testAccStageConfigTags1(rName, "key2", "value2"),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "tags.%", "1"),
+					resource.TestCheckResourceAttr(resourceName, "tags.key2", "value2"),
+				),
+			},
+		},
+	})
+}
+
 func TestAccAPIGatewayStage_disappears(t *testing.T) {
 	var stage apigateway.Stage
 	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
@@ -125,6 +220,31 @@ func TestAccAPIGatewayStage_disappears(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &stage),
 					acctest.CheckResourceDisappears(acctest.Provider, tfapigateway.ResourceStage(), resourceName),
+					acctest.CheckResourceDisappears(acctest.Provider, tfapigateway.ResourceStage(), resourceName),
+				),
+				ExpectNonEmptyPlan: true,
+			},
+		},
+	})
+}
+
+func TestAccAPIGatewayStage_disappears_restApi(t *testing.T) {
+	var stage apigateway.Stage
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	resourceName := "aws_api_gateway_stage.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t); acctest.PreCheckAPIGatewayTypeEDGE(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, apigateway.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckStageDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccStageReferencingDeploymentConfig(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &stage),
+					acctest.CheckResourceDisappears(acctest.Provider, tfapigateway.ResourceRestAPI(), "aws_api_gateway_rest_api.test"),
+					acctest.CheckResourceDisappears(acctest.Provider, tfapigateway.ResourceStage(), resourceName),
 				),
 				ExpectNonEmptyPlan: true,
 			},
@@ -190,7 +310,7 @@ func TestAccAPIGatewayStage_accessLogSettings(t *testing.T) {
 				),
 			},
 			{
-				Config: testAccStageConfig_basic(rName),
+				Config: testAccStageConfigBasic(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
@@ -205,6 +325,7 @@ func TestAccAPIGatewayStage_AccessLogSettings_kinesis(t *testing.T) {
 	var conf apigateway.Stage
 	rName := sdkacctest.RandString(5)
 	resourceName := "aws_api_gateway_stage.test"
+	kinesesResourceName := "aws_kinesis_firehose_delivery_stream.test"
 	clf := `$context.identity.sourceIp $context.identity.caller $context.identity.user [$context.requestTime] "$context.httpMethod $context.resourcePath $context.protocol" $context.status $context.responseLength $context.requestId`
 	json := `{ "requestId":"$context.requestId", "ip": "$context.identity.sourceIp", "caller":"$context.identity.caller", "user":"$context.identity.user", "requestTime":"$context.requestTime", "httpMethod":"$context.httpMethod", "resourcePath":"$context.resourcePath", "status":"$context.status", "protocol":"$context.protocol", "responseLength":"$context.responseLength" }`
 	xml := `<request id="$context.requestId"> <ip>$context.identity.sourceIp</ip> <caller>$context.identity.caller</caller> <user>$context.identity.user</user> <requestTime>$context.requestTime</requestTime> <httpMethod>$context.httpMethod</httpMethod> <resourcePath>$context.resourcePath</resourcePath> <status>$context.status</status> <protocol>$context.protocol</protocol> <responseLength>$context.responseLength</responseLength> </request>`
@@ -221,7 +342,7 @@ func TestAccAPIGatewayStage_AccessLogSettings_kinesis(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &conf),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.#", "1"),
-					acctest.MatchResourceAttrRegionalARN(resourceName, "access_log_settings.0.destination_arn", "firehose", regexp.MustCompile(`deliverystream/amazon-apigateway-.+`)),
+					resource.TestCheckResourceAttrPair(resourceName, "access_log_settings.0.destination_arn", kinesesResourceName, "arn"),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.0.format", clf),
 				),
 			},
@@ -232,7 +353,7 @@ func TestAccAPIGatewayStage_AccessLogSettings_kinesis(t *testing.T) {
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.#", "1"),
-					acctest.MatchResourceAttrRegionalARN(resourceName, "access_log_settings.0.destination_arn", "firehose", regexp.MustCompile(`deliverystream/amazon-apigateway-.+`)),
+					resource.TestCheckResourceAttrPair(resourceName, "access_log_settings.0.destination_arn", kinesesResourceName, "arn"),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.0.format", json),
 				),
 			},
@@ -242,7 +363,7 @@ func TestAccAPIGatewayStage_AccessLogSettings_kinesis(t *testing.T) {
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.#", "1"),
-					acctest.MatchResourceAttrRegionalARN(resourceName, "access_log_settings.0.destination_arn", "firehose", regexp.MustCompile(`deliverystream/amazon-apigateway-.+`)),
+					resource.TestCheckResourceAttrPair(resourceName, "access_log_settings.0.destination_arn", kinesesResourceName, "arn"),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.0.format", xml),
 				),
 			},
@@ -252,12 +373,12 @@ func TestAccAPIGatewayStage_AccessLogSettings_kinesis(t *testing.T) {
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.#", "1"),
-					acctest.MatchResourceAttrRegionalARN(resourceName, "access_log_settings.0.destination_arn", "firehose", regexp.MustCompile(`deliverystream/amazon-apigateway-.+`)),
+					resource.TestCheckResourceAttrPair(resourceName, "access_log_settings.0.destination_arn", kinesesResourceName, "arn"),
 					resource.TestCheckResourceAttr(resourceName, "access_log_settings.0.format", csv),
 				),
 			},
 			{
-				Config: testAccStageConfig_basic(rName),
+				Config: testAccStageConfigBasic(rName),
 				Check: resource.ComposeTestCheckFunc(
 					testAccCheckStageExists(resourceName, &conf),
 					acctest.MatchResourceAttrRegionalARNNoAccount(resourceName, "arn", "apigateway", regexp.MustCompile(`/restapis/.+/stages/prod`)),
@@ -268,6 +389,40 @@ func TestAccAPIGatewayStage_AccessLogSettings_kinesis(t *testing.T) {
 	})
 }
 
+func TestAccAPIGatewayStage_waf(t *testing.T) {
+	var conf apigateway.Stage
+	rName := sdkacctest.RandString(5)
+	resourceName := "aws_api_gateway_stage.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { acctest.PreCheck(t); acctest.PreCheckAPIGatewayTypeEDGE(t) },
+		ErrorCheck:   acctest.ErrorCheck(t, apigateway.EndpointsID),
+		Providers:    acctest.Providers,
+		CheckDestroy: testAccCheckStageDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccAWSAPIGatewayStageConfigWafAcl(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+				),
+			},
+			{
+				Config: testAccAWSAPIGatewayStageConfigWafAcl(rName),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckStageExists(resourceName, &conf),
+					resource.TestCheckResourceAttrPair(resourceName, "web_acl_arn", "aws_wafregional_web_acl.test", "arn"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateIdFunc: testAccStageImportStateIdFunc(resourceName),
+				ImportStateVerify: true,
+			},
+		},
+	})
+}
+
 func testAccCheckStageExists(n string, res *apigateway.Stage) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		rs, ok := s.RootModule().Resources[n]
@@ -280,16 +435,15 @@ func testAccCheckStageExists(n string, res *apigateway.Stage) resource.TestCheck
 		}
 
 		conn := acctest.Provider.Meta().(*conns.AWSClient).APIGatewayConn
-
-		req := &apigateway.GetStageInput{
-			RestApiId: aws.String(s.RootModule().Resources["aws_api_gateway_rest_api.test"].Primary.ID),
-			StageName: aws.String(rs.Primary.Attributes["stage_name"]),
-		}
-		out, err := conn.GetStage(req)
+		out, err := tfapigateway.FindStageByName(conn, rs.Primary.Attributes["rest_api_id"], rs.Primary.Attributes["stage_name"])
 		if err != nil {
 			return err
 		}
 
+		if out == nil {
+			return fmt.Errorf("API Gateway Stage not found")
+		}
+
 		*res = *out
 
 		return nil
@@ -304,24 +458,16 @@ func testAccCheckStageDestroy(s *terraform.State) error {
 			continue
 		}
 
-		req := &apigateway.GetStageInput{
-			RestApiId: aws.String(s.RootModule().Resources["aws_api_gateway_rest_api.test"].Primary.ID),
-			StageName: aws.String(rs.Primary.Attributes["stage_name"]),
-		}
-		out, err := conn.GetStage(req)
-		if err == nil {
-			return fmt.Errorf("API Gateway Stage still exists: %s", out)
+		_, err := tfapigateway.FindStageByName(conn, rs.Primary.Attributes["rest_api_id"], rs.Primary.Attributes["stage_name"])
+		if tfresource.NotFound(err) {
+			continue
 		}
 
-		awsErr, ok := err.(awserr.Error)
-		if !ok {
-			return err
-		}
-		if awsErr.Code() != apigateway.ErrCodeNotFoundException {
+		if err != nil {
 			return err
 		}
 
-		return nil
+		return fmt.Errorf("API Gateway Stage %s still exists", rs.Primary.ID)
 	}
 
 	return nil
@@ -471,47 +617,45 @@ resource "aws_api_gateway_deployment" "test2" {
 `)
 }
 
-func testAccStageConfig_basic(rName string) string {
+func testAccStageConfigBasic(rName string) string {
 	return testAccStageConfig_base(rName) + `
 resource "aws_api_gateway_stage" "test" {
-  rest_api_id           = aws_api_gateway_rest_api.test.id
-  stage_name            = "prod"
-  deployment_id         = aws_api_gateway_deployment.dev.id
-  cache_cluster_enabled = true
-  cache_cluster_size    = "0.5"
-  xray_tracing_enabled  = true
-  variables = {
-    one = "1"
-    two = "2"
-  }
-  tags = {
-    Name = "tf-test"
-  }
+  rest_api_id   = aws_api_gateway_rest_api.test.id
+  stage_name    = "prod"
+  deployment_id = aws_api_gateway_deployment.dev.id
 }
 `
 }
 
-func testAccStageConfig_updated(rName string) string {
+func testAccStageConfigUpdated(rName string) string {
 	return testAccStageConfig_base(rName) + `
 resource "aws_api_gateway_stage" "test" {
-  rest_api_id           = aws_api_gateway_rest_api.test.id
-  stage_name            = "prod"
-  deployment_id         = aws_api_gateway_deployment.dev.id
-  cache_cluster_enabled = false
-  description           = "Hello world"
-  xray_tracing_enabled  = false
+  rest_api_id          = aws_api_gateway_rest_api.test.id
+  stage_name           = "prod"
+  deployment_id        = aws_api_gateway_deployment.dev.id
+  description          = "Hello world"
+  xray_tracing_enabled = true
+
   variables = {
     one   = "1"
     three = "3"
   }
-  tags = {
-    Name      = "tf-test"
-    ExtraName = "tf-test"
-  }
 }
 `
 }
 
+func testAccStageConfigCacheConfig(rName, size string) string {
+	return testAccStageConfig_base(rName) + fmt.Sprintf(`
+resource "aws_api_gateway_stage" "test" {
+  rest_api_id           = aws_api_gateway_rest_api.test.id
+  stage_name            = "prod"
+  deployment_id         = aws_api_gateway_deployment.dev.id
+  cache_cluster_enabled = true
+  cache_cluster_size    = %[1]q
+}
+`, size)
+}
+
 func testAccStageConfig_accessLogSettings(rName string, format string) string {
 	return testAccStageConfig_base(rName) + fmt.Sprintf(`
 resource "aws_cloudwatch_log_group" "test" {
@@ -519,18 +663,10 @@ resource "aws_cloudwatch_log_group" "test" {
 }
 
 resource "aws_api_gateway_stage" "test" {
-  rest_api_id           = aws_api_gateway_rest_api.test.id
-  stage_name            = "prod"
-  deployment_id         = aws_api_gateway_deployment.dev.id
-  cache_cluster_enabled = true
-  cache_cluster_size    = "0.5"
-  variables = {
-    one = "1"
-    two = "2"
-  }
-  tags = {
-    Name = "tf-test"
-  }
+  rest_api_id   = aws_api_gateway_rest_api.test.id
+  stage_name    = "prod"
+  deployment_id = aws_api_gateway_deployment.dev.id
+
   access_log_settings {
     destination_arn = aws_cloudwatch_log_group.test.arn
     format          = %q
@@ -577,18 +713,10 @@ resource "aws_kinesis_firehose_delivery_stream" "test" {
 }
 
 resource "aws_api_gateway_stage" "test" {
-  rest_api_id           = aws_api_gateway_rest_api.test.id
-  stage_name            = "prod"
-  deployment_id         = aws_api_gateway_deployment.dev.id
-  cache_cluster_enabled = true
-  cache_cluster_size    = "0.5"
-  variables = {
-    one = "1"
-    two = "2"
-  }
-  tags = {
-    Name = "tf-test"
-  }
+  rest_api_id   = aws_api_gateway_rest_api.test.id
+  stage_name    = "prod"
+  deployment_id = aws_api_gateway_deployment.dev.id
+
   access_log_settings {
     destination_arn = aws_kinesis_firehose_delivery_stream.test.arn
     format          = %q
@@ -596,3 +724,49 @@ resource "aws_api_gateway_stage" "test" {
 }
 `, rName, format)
 }
+
+func testAccStageConfigTags1(rName, tagKey1, tagValue1 string) string {
+	return testAccStageConfig_base(rName) + fmt.Sprintf(`
+resource "aws_api_gateway_stage" "test" {
+  rest_api_id   = aws_api_gateway_rest_api.test.id
+  stage_name    = "prod"
+  deployment_id = aws_api_gateway_deployment.dev.id
+
+  tags = {
+    %[1]q = %[2]q
+  }
+}
+`, tagKey1, tagValue1)
+}
+
+func testAccStageConfigTags2(rName, tagKey1, tagValue1, tagKey2, tagValue2 string) string {
+	return testAccStageConfig_base(rName) + fmt.Sprintf(`
+resource "aws_api_gateway_stage" "test" {
+  rest_api_id   = aws_api_gateway_rest_api.test.id
+  stage_name    = "prod"
+  deployment_id = aws_api_gateway_deployment.dev.id
+
+  tags = {
+    %[1]q = %[2]q
+    %[3]q = %[4]q
+  }
+}
+`, tagKey1, tagValue1, tagKey2, tagValue2)
+}
+
+func testAccAWSAPIGatewayStageConfigWafAcl(rName string) string {
+	return testAccStageConfigBasic(rName) + fmt.Sprintf(`
+resource "aws_wafregional_web_acl" "test" {
+  name        = %[1]q
+  metric_name = "test"
+  default_action {
+    type = "ALLOW"
+  }
+}
+
+resource "aws_wafregional_web_acl_association" "test" {
+  resource_arn = aws_api_gateway_stage.test.arn
+  web_acl_id   = aws_wafregional_web_acl.test.id
+}
+`, rName)
+}
diff --git a/internal/service/apigateway/status.go b/internal/service/apigateway/status.go
index 4ae4f5142ed..87514525f4d 100644
--- a/internal/service/apigateway/status.go
+++ b/internal/service/apigateway/status.go
@@ -7,6 +7,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/apigateway"
 	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
 func apiGatewayVpcLinkStatus(conn *apigateway.APIGateway, vpcLinkId string) resource.StateRefreshFunc {
@@ -29,3 +30,18 @@ func apiGatewayVpcLinkStatus(conn *apigateway.APIGateway, vpcLinkId string) reso
 		return output, aws.StringValue(output.Status), nil
 	}
 }
+
+func stageCacheStatus(conn *apigateway.APIGateway, restApiId, name string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		output, err := FindStageByName(conn, restApiId, name)
+
+		if tfresource.NotFound(err) {
+			return nil, "", nil
+		}
+		if err != nil {
+			return nil, "", err
+		}
+
+		return output, aws.StringValue(output.CacheClusterStatus), nil
+	}
+}
diff --git a/internal/service/apigateway/wait.go b/internal/service/apigateway/wait.go
index 817acae9678..88387a58324 100644
--- a/internal/service/apigateway/wait.go
+++ b/internal/service/apigateway/wait.go
@@ -13,6 +13,12 @@ const (
 
 	// Maximum amount of time for VpcLink to delete
 	apiGatewayVPCLinkDeleteTimeout = 20 * time.Minute
+
+	// Maximum amount of time for Stage Cache to be available
+	apiGatewayStageCacheAvailableTimeout = 90 * time.Minute
+
+	// Maximum amount of time for Stage Cache to update
+	apiGatewayStageCacheUpdateTimeout = 30 * time.Minute
 )
 
 func waitAPIGatewayVPCLinkAvailable(conn *apigateway.APIGateway, vpcLinkId string) error {
@@ -46,3 +52,49 @@ func waitAPIGatewayVPCLinkDeleted(conn *apigateway.APIGateway, vpcLinkId string)
 
 	return err
 }
+
+func waitStageCacheAvailable(conn *apigateway.APIGateway, restApiId, name string) (*apigateway.Stage, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{
+			apigateway.CacheClusterStatusCreateInProgress,
+			apigateway.CacheClusterStatusDeleteInProgress,
+			apigateway.CacheClusterStatusFlushInProgress,
+		},
+		Target:  []string{apigateway.CacheClusterStatusAvailable},
+		Refresh: stageCacheStatus(conn, restApiId, name),
+		Timeout: apiGatewayStageCacheAvailableTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if output, ok := outputRaw.(*apigateway.Stage); ok {
+		return output, err
+	}
+
+	return nil, err
+}
+
+func waitStageCacheUpdated(conn *apigateway.APIGateway, restApiId, name string) (*apigateway.Stage, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{
+			apigateway.CacheClusterStatusCreateInProgress,
+			apigateway.CacheClusterStatusFlushInProgress,
+		},
+		Target: []string{
+			apigateway.CacheClusterStatusAvailable,
+			// There's an AWS API bug (raised & confirmed in Sep 2016 by support)
+			// which causes the stage to remain in deletion state forever
+			apigateway.CacheClusterStatusDeleteInProgress,
+		},
+		Refresh: stageCacheStatus(conn, restApiId, name),
+		Timeout: apiGatewayStageCacheUpdateTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if output, ok := outputRaw.(*apigateway.Stage); ok {
+		return output, err
+	}
+
+	return nil, err
+}
diff --git a/website/docs/r/api_gateway_stage.html.markdown b/website/docs/r/api_gateway_stage.html.markdown
index 120e0024d8e..5954fd17393 100644
--- a/website/docs/r/api_gateway_stage.html.markdown
+++ b/website/docs/r/api_gateway_stage.html.markdown
@@ -136,6 +136,7 @@ In addition to all arguments above, the following attributes are exported:
   when allowing API Gateway to invoke a Lambda function,
   e.g., `arn:aws:execute-api:eu-west-2:123456789012:z4675bid1j/prod`
 * `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).
+* `web_acl_arn` - The ARN of the WebAcl associated with the Stage.
 
 ## Import