Merge pull request #16005 from ewbankkit/b-aws_kinesis_analyltics_app…

…lication-add-and-remove-input-processing_configuration

r/aws_kinesis_analytics_application: Correct resource update

aws/internal/service/kinesisanalytics/finder/finder.go

@@ -0,0 +1,20 @@
+package finder
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/kinesisanalytics"
+)
+
+// ApplicationByName returns the application corresponding to the specified name.
+func ApplicationByName(conn *kinesisanalytics.KinesisAnalytics, name string) (*kinesisanalytics.ApplicationDetail, error) {
+	input := &kinesisanalytics.DescribeApplicationInput{
+		ApplicationName: aws.String(name),
+	}
+
+	output, err := conn.DescribeApplication(input)
+	if err != nil {
+		return nil, err
+	}
+
+	return output.ApplicationDetail, nil
+}

aws/internal/service/kinesisanalytics/waiter/status.go

@@ -3,34 +3,27 @@ package waiter
 import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/kinesisanalytics"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/service/kinesisanalytics/finder"
 )
 
 const (
-	// ApplicationStatus NotFound
-	ApplicationStatusNotFound = "NotFound"
-
-	// ApplicationStatus Unknown
-	ApplicationStatusUnknown = "Unknown"
+	applicationStatusNotFound = "NotFound"
+	applicationStatusUnknown  = "Unknown"
 )
 
 // ApplicationStatus fetches the Application and its Status
-func ApplicationStatus(conn *kinesisanalytics.KinesisAnalytics, applicationName string) resource.StateRefreshFunc {
+func ApplicationStatus(conn *kinesisanalytics.KinesisAnalytics, name string) resource.StateRefreshFunc {
 	return func() (interface{}, string, error) {
-		input := &kinesisanalytics.DescribeApplicationInput{
-			ApplicationName: aws.String(applicationName),
-		}
-
-		output, err := conn.DescribeApplication(input)
+		application, err := finder.ApplicationByName(conn, name)
 
-		if err != nil {
-			return nil, ApplicationStatusUnknown, err
+		if tfawserr.ErrCodeEquals(err, kinesisanalytics.ErrCodeResourceNotFoundException) {
+			return nil, applicationStatusNotFound, nil
 		}
 
-		application := output.ApplicationDetail
-
-		if application == nil {
-			return application, ApplicationStatusNotFound, nil
+		if err != nil {
+			return nil, applicationStatusUnknown, err
 		}
 
 		return application, aws.StringValue(application.ApplicationStatus), nil

aws/internal/service/kinesisanalytics/waiter/waiter.go

@@ -4,21 +4,19 @@ import (
 	"time"
 
 	"github.com/aws/aws-sdk-go/service/kinesisanalytics"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	iamwaiter "github.com/terraform-providers/terraform-provider-aws/aws/internal/service/iam/waiter"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/tfresource"
 )
 
-const (
-	// Maximum amount of time to wait for an Application to be deleted
-	ApplicationDeletedTimeout = 20 * time.Minute
-)
-
-// ApplicationDeleted waits for an Application to be deleted
-func ApplicationDeleted(conn *kinesisanalytics.KinesisAnalytics, applicationName string) (*kinesisanalytics.ApplicationSummary, error) {
+// ApplicationDeleted waits for an Application to return Deleted
+func ApplicationDeleted(conn *kinesisanalytics.KinesisAnalytics, name string, timeout time.Duration) (*kinesisanalytics.ApplicationSummary, error) {
 	stateConf := &resource.StateChangeConf{
 		Pending: []string{kinesisanalytics.ApplicationStatusRunning, kinesisanalytics.ApplicationStatusDeleting},
-		Target:  []string{ApplicationStatusNotFound},
-		Refresh: ApplicationStatus(conn, applicationName),
-		Timeout: ApplicationDeletedTimeout,
+		Target:  []string{},
+		Refresh: ApplicationStatus(conn, name),
+		Timeout: timeout,
 	}
 
 	outputRaw, err := stateConf.WaitForState()
@@ -29,3 +27,46 @@ func ApplicationDeleted(conn *kinesisanalytics.KinesisAnalytics, applicationName
 
 	return nil, err
 }
+
+// IAMPropagation retries the specified function if the returned error indicates an IAM eventual consistency issue.
+// If the retries time out the specified function is called one last time.
+func IAMPropagation(f func() (interface{}, error)) (interface{}, error) {
+	var output interface{}
+
+	err := resource.Retry(iamwaiter.PropagationTimeout, func() *resource.RetryError {
+		var err error
+
+		output, err = f()
+
+		// Kinesis Stream: https://github.com/hashicorp/terraform-provider-aws/issues/7032
+		if tfawserr.ErrMessageContains(err, kinesisanalytics.ErrCodeInvalidArgumentException, "Kinesis Analytics service doesn't have sufficient privileges") {
+			return resource.RetryableError(err)
+		}
+
+		// Kinesis Firehose: https://github.com/hashicorp/terraform-provider-aws/issues/7394
+		if tfawserr.ErrMessageContains(err, kinesisanalytics.ErrCodeInvalidArgumentException, "Kinesis Analytics doesn't have sufficient privileges") {
+			return resource.RetryableError(err)
+		}
+
+		// InvalidArgumentException: Given IAM role arn : arn:aws:iam::123456789012:role/xxx does not provide Invoke permissions on the Lambda resource : arn:aws:lambda:us-west-2:123456789012:function:yyy
+		if tfawserr.ErrMessageContains(err, kinesisanalytics.ErrCodeInvalidArgumentException, "does not provide Invoke permissions on the Lambda resource") {
+			return resource.RetryableError(err)
+		}
+
+		if err != nil {
+			return resource.NonRetryableError(err)
+		}
+
+		return nil
+	})
+
+	if tfresource.TimedOut(err) {
+		output, err = f()
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return output, nil
+}