Merge pull request #18056 from ewbankkit/f-r/aws_kinesisanalyticsv2_a…

…pplication-start/stop

Start/stop Kinesis Analytics Flink application and add application snapshot resource

.changelog/18056.txt

@@ -0,0 +1,19 @@
+```release-note:enhancement
+resource/aws_kinesisanalyticsv2_application: Add `start_application` attribute
+```
+
+```release-note:enhancement
+resource/aws_kinesisanalyticsv2_application: `starting_position_configuration` can be specified when starting a SQL application
+```
+
+```release-note:enhancement
+resource/aws_kinesisanalyticsv2_application: Add `run_configuration` attribute for starting a Flink application
+```
+
+```release-note:enhancement
+resource/aws_kinesisanalyticsv2_application: Add `force_stop` attribute
+```
+
+```release-note:new-resource
+aws_kinesisanalyticsv2_application_snapshot
+```

aws/internal/service/kinesisanalyticsv2/finder/finder.go

@@ -3,18 +3,86 @@ package finder
 import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/kinesisanalyticsv2"
+	"github.com/hashicorp/aws-sdk-go-base/tfawserr"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
 )
 
-// ApplicationByName returns the application corresponding to the specified name.
-func ApplicationByName(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) (*kinesisanalyticsv2.ApplicationDetail, error) {
+// ApplicationDetailByName returns the application corresponding to the specified name.
+// Returns NotFoundError if no application is found.
+func ApplicationDetailByName(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) (*kinesisanalyticsv2.ApplicationDetail, error) {
 	input := &kinesisanalyticsv2.DescribeApplicationInput{
 		ApplicationName: aws.String(name),
 	}
 
+	return ApplicationDetail(conn, input)
+}
+
+// ApplicationDetail returns the application details corresponding to the specified input.
+// Returns NotFoundError if no application is found.
+func ApplicationDetail(conn *kinesisanalyticsv2.KinesisAnalyticsV2, input *kinesisanalyticsv2.DescribeApplicationInput) (*kinesisanalyticsv2.ApplicationDetail, error) {
 	output, err := conn.DescribeApplication(input)
+
+	if tfawserr.ErrCodeEquals(err, kinesisanalyticsv2.ErrCodeResourceNotFoundException) {
+		return nil, &resource.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
 	if err != nil {
 		return nil, err
 	}
 
+	if output == nil || output.ApplicationDetail == nil {
+		return nil, &resource.NotFoundError{
+			Message:     "Empty result",
+			LastRequest: input,
+		}
+	}
+
 	return output.ApplicationDetail, nil
 }
+
+// SnapshotDetailsByApplicationAndSnapshotNames returns the application snapshot details corresponding to the specified application and snapshot names.
+// Returns NotFoundError if no application snapshot is found.
+func SnapshotDetailsByApplicationAndSnapshotNames(conn *kinesisanalyticsv2.KinesisAnalyticsV2, applicationName, snapshotName string) (*kinesisanalyticsv2.SnapshotDetails, error) {
+	input := &kinesisanalyticsv2.DescribeApplicationSnapshotInput{
+		ApplicationName: aws.String(applicationName),
+		SnapshotName:    aws.String(snapshotName),
+	}
+
+	return SnapshotDetails(conn, input)
+}
+
+// SnapshotDetails returns the application snapshot details corresponding to the specified input.
+// Returns NotFoundError if no application snapshot is found.
+func SnapshotDetails(conn *kinesisanalyticsv2.KinesisAnalyticsV2, input *kinesisanalyticsv2.DescribeApplicationSnapshotInput) (*kinesisanalyticsv2.SnapshotDetails, error) {
+	output, err := conn.DescribeApplicationSnapshot(input)
+
+	if tfawserr.ErrCodeEquals(err, kinesisanalyticsv2.ErrCodeResourceNotFoundException) {
+		return nil, &resource.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if tfawserr.ErrMessageContains(err, kinesisanalyticsv2.ErrCodeInvalidArgumentException, "does not exist") {
+		return nil, &resource.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	if output == nil || output.SnapshotDetails == nil {
+		return nil, &resource.NotFoundError{
+			Message:     "Empty result",
+			LastRequest: input,
+		}
+	}
+
+	return output.SnapshotDetails, nil
+}

aws/internal/service/kinesisanalyticsv2/id.go

@@ -0,0 +1,25 @@
+package kinesisanalyticsv2
+
+import (
+	"fmt"
+	"strings"
+)
+
+const applicationSnapshotIDSeparator = "/"
+
+func ApplicationSnapshotCreateID(applicationName, snapshotName string) string {
+	parts := []string{applicationName, snapshotName}
+	id := strings.Join(parts, applicationSnapshotIDSeparator)
+
+	return id
+}
+
+func ApplicationSnapshotParseID(id string) (string, string, error) {
+	parts := strings.Split(id, applicationSnapshotIDSeparator)
+
+	if len(parts) == 2 && parts[0] != "" && parts[1] != "" {
+		return parts[0], parts[1], nil
+	}
+
+	return "", "", fmt.Errorf("unexpected format for ID (%q), expected application-name%ssnapshot-name", id, applicationSnapshotIDSeparator)
+}

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

@@ -3,29 +3,41 @@ package waiter
 import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/kinesisanalyticsv2"
-	"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/kinesisanalyticsv2/finder"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/tfresource"
 )
 
-const (
-	applicationStatusNotFound = "NotFound"
-	applicationStatusUnknown  = "Unknown"
-)
-
-// ApplicationStatus fetches the Application and its Status
+// ApplicationStatus fetches the ApplicationDetail and its Status
 func ApplicationStatus(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) resource.StateRefreshFunc {
 	return func() (interface{}, string, error) {
-		application, err := finder.ApplicationByName(conn, name)
+		applicationDetail, err := finder.ApplicationDetailByName(conn, name)
+
+		if tfresource.NotFound(err) {
+			return nil, "", nil
+		}
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		return applicationDetail, aws.StringValue(applicationDetail.ApplicationStatus), nil
+	}
+}
+
+// SnapshotDetailsStatus fetches the SnapshotDetails and its Status
+func SnapshotDetailsStatus(conn *kinesisanalyticsv2.KinesisAnalyticsV2, applicationName, snapshotName string) resource.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		snapshotDetails, err := finder.SnapshotDetailsByApplicationAndSnapshotNames(conn, applicationName, snapshotName)
 
-		if tfawserr.ErrCodeEquals(err, kinesisanalyticsv2.ErrCodeResourceNotFoundException) {
-			return nil, applicationStatusNotFound, nil
+		if tfresource.NotFound(err) {
+			return nil, "", nil
 		}
 
 		if err != nil {
-			return nil, applicationStatusUnknown, err
+			return nil, "", err
 		}
 
-		return application, aws.StringValue(application.ApplicationStatus), nil
+		return snapshotDetails, aws.StringValue(snapshotDetails.SnapshotStatus), nil
 	}
 }

aws/internal/service/kinesisanalyticsv2/waiter/waiter.go

@@ -10,13 +10,77 @@ import (
 	"github.com/terraform-providers/terraform-provider-aws/aws/internal/tfresource"
 )
 
+const (
+	ApplicationDeletedTimeout = 5 * time.Minute
+	ApplicationStartedTimeout = 5 * time.Minute
+	ApplicationStoppedTimeout = 5 * time.Minute
+	ApplicationUpdatedTimeout = 5 * time.Minute
+
+	SnapshotCreatedTimeout = 5 * time.Minute
+	SnapshotDeletedTimeout = 5 * time.Minute
+)
+
 // ApplicationDeleted waits for an Application to return Deleted
-func ApplicationDeleted(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string, timeout time.Duration) (*kinesisanalyticsv2.ApplicationDetail, error) {
+func ApplicationDeleted(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) (*kinesisanalyticsv2.ApplicationDetail, error) {
 	stateConf := &resource.StateChangeConf{
 		Pending: []string{kinesisanalyticsv2.ApplicationStatusDeleting},
 		Target:  []string{},
 		Refresh: ApplicationStatus(conn, name),
-		Timeout: timeout,
+		Timeout: ApplicationDeletedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if v, ok := outputRaw.(*kinesisanalyticsv2.ApplicationDetail); ok {
+		return v, err
+	}
+
+	return nil, err
+}
+
+// ApplicationStarted waits for an Application to start
+func ApplicationStarted(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) (*kinesisanalyticsv2.ApplicationDetail, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{kinesisanalyticsv2.ApplicationStatusStarting},
+		Target:  []string{kinesisanalyticsv2.ApplicationStatusRunning},
+		Refresh: ApplicationStatus(conn, name),
+		Timeout: ApplicationStartedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if v, ok := outputRaw.(*kinesisanalyticsv2.ApplicationDetail); ok {
+		return v, err
+	}
+
+	return nil, err
+}
+
+// ApplicationStopped waits for an Application to stop
+func ApplicationStopped(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) (*kinesisanalyticsv2.ApplicationDetail, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{kinesisanalyticsv2.ApplicationStatusForceStopping, kinesisanalyticsv2.ApplicationStatusStopping},
+		Target:  []string{kinesisanalyticsv2.ApplicationStatusReady},
+		Refresh: ApplicationStatus(conn, name),
+		Timeout: ApplicationStoppedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if v, ok := outputRaw.(*kinesisanalyticsv2.ApplicationDetail); ok {
+		return v, err
+	}
+
+	return nil, err
+}
+
+// ApplicationUpdated waits for an Application to return Deleted
+func ApplicationUpdated(conn *kinesisanalyticsv2.KinesisAnalyticsV2, name string) (*kinesisanalyticsv2.ApplicationDetail, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{kinesisanalyticsv2.ApplicationStatusUpdating},
+		Target:  []string{kinesisanalyticsv2.ApplicationStatusReady, kinesisanalyticsv2.ApplicationStatusRunning},
+		Refresh: ApplicationStatus(conn, name),
+		Timeout: ApplicationUpdatedTimeout,
 	}
 
 	outputRaw, err := stateConf.WaitForState()
@@ -75,3 +139,39 @@ func IAMPropagation(f func() (interface{}, error)) (interface{}, error) {
 
 	return output, nil
 }
+
+// SnapshotCreated waits for a Snapshot to return Created
+func SnapshotCreated(conn *kinesisanalyticsv2.KinesisAnalyticsV2, applicationName, snapshotName string) (*kinesisanalyticsv2.SnapshotDetails, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{kinesisanalyticsv2.SnapshotStatusCreating},
+		Target:  []string{kinesisanalyticsv2.SnapshotStatusReady},
+		Refresh: SnapshotDetailsStatus(conn, applicationName, snapshotName),
+		Timeout: SnapshotCreatedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if v, ok := outputRaw.(*kinesisanalyticsv2.SnapshotDetails); ok {
+		return v, err
+	}
+
+	return nil, err
+}
+
+// SnapshotDeleted waits for a Snapshot to return Deleted
+func SnapshotDeleted(conn *kinesisanalyticsv2.KinesisAnalyticsV2, applicationName, snapshotName string) (*kinesisanalyticsv2.SnapshotDetails, error) {
+	stateConf := &resource.StateChangeConf{
+		Pending: []string{kinesisanalyticsv2.SnapshotStatusDeleting},
+		Target:  []string{},
+		Refresh: SnapshotDetailsStatus(conn, applicationName, snapshotName),
+		Timeout: SnapshotDeletedTimeout,
+	}
+
+	outputRaw, err := stateConf.WaitForState()
+
+	if v, ok := outputRaw.(*kinesisanalyticsv2.SnapshotDetails); ok {
+		return v, err
+	}
+
+	return nil, err
+}

aws/provider.go

@@ -792,6 +792,7 @@ func Provider() *schema.Provider {
 			"aws_key_pair":                                            resourceAwsKeyPair(),
 			"aws_kinesis_analytics_application":                       resourceAwsKinesisAnalyticsApplication(),
 			"aws_kinesisanalyticsv2_application":                      resourceAwsKinesisAnalyticsV2Application(),
+			"aws_kinesisanalyticsv2_application_snapshot":             resourceAwsKinesisAnalyticsV2ApplicationSnapshot(),
 			"aws_kinesis_firehose_delivery_stream":                    resourceAwsKinesisFirehoseDeliveryStream(),
 			"aws_kinesis_stream":                                      resourceAwsKinesisStream(),
 			"aws_kinesis_stream_consumer":                             resourceAwsKinesisStreamConsumer(),