resource/aws_cloudwatch_log_subscription_filter: Perform eventual con…

…sistency retries on update (#11739)

Previously in the acceptance testing:

```
--- FAIL: TestAccAWSCloudwatchLogSubscriptionFilter_RoleArn (81.73s)
    testing.go:640: Step 2 error: errors during apply:

        Error: Error updating SubscriptionFilter (tf-acc-test-2641008305567738726) for LogGroup (tf-acc-test-2641008305567738726), message: "Could not deliver test message to specified Kinesis stream. Check if the given kinesis stream is in ACTIVE state.", code: "InvalidParameterException"
```

Previously, the resource was only performing eventual consistency retries on creation. Since arguments that are subject to the same eventual consistency issues are updateable, this allows the resource to handle situations where the configuration update should be valid.

Output from acceptance testing:

```
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_disappears (38.37s)
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_basic (43.35s)
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_disappears_LogGroup (50.76s)
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_Distribution (71.23s)
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_DestinationArn_KinesisStream (82.84s)
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_DestinationArn_KinesisDataFirehose (107.36s)
--- PASS: TestAccAWSCloudwatchLogSubscriptionFilter_RoleArn (118.09s)
```

aws/resource_aws_cloudwatch_log_subscription_filter.go

@@ -99,13 +99,28 @@ func resourceAwsCloudwatchLogSubscriptionFilterUpdate(d *schema.ResourceData, me
 	params := getAwsCloudWatchLogsSubscriptionFilterInput(d)
 
 	log.Printf("[DEBUG] Update SubscriptionFilter %#v", params)
-	_, err := conn.PutSubscriptionFilter(&params)
-	if err != nil {
-		if awsErr, ok := err.(awserr.Error); ok {
-			return fmt.Errorf("Error updating SubscriptionFilter (%s) for LogGroup (%s), message: \"%s\", code: \"%s\"",
-				d.Get("name").(string), d.Get("log_group_name").(string), awsErr.Message(), awsErr.Code())
+
+	err := resource.Retry(5*time.Minute, func() *resource.RetryError {
+		_, err := conn.PutSubscriptionFilter(&params)
+
+		if isAWSErr(err, cloudwatchlogs.ErrCodeInvalidParameterException, "Could not deliver test message to specified") {
+			return resource.RetryableError(err)
 		}
-		return err
+		if isAWSErr(err, cloudwatchlogs.ErrCodeInvalidParameterException, "Could not execute the lambda function") {
+			return resource.RetryableError(err)
+		}
+		if err != nil {
+			return resource.NonRetryableError(err)
+		}
+		return nil
+	})
+
+	if isResourceTimeoutError(err) {
+		_, err = conn.PutSubscriptionFilter(&params)
+	}
+
+	if err != nil {
+		return fmt.Errorf("error updating CloudWatch Log Subscription Filter (%s): %w", d.Get("log_group_name").(string), err)
 	}
 
 	d.SetId(cloudwatchLogsSubscriptionFilterId(d.Get("log_group_name").(string)))