From 3ce9adff4ce49ff58550f188dafc594bd5947c14 Mon Sep 17 00:00:00 2001
From: Dirk Avery <dirk.avery@gmail.com>
Date: Wed, 7 Apr 2021 13:38:44 -0400
Subject: [PATCH] tests/r/dms_replication_instance: Add sweeper concurrency

---
 ...ource_aws_dms_replication_instance_test.go | 58 ++++++++-----------
 1 file changed, 23 insertions(+), 35 deletions(-)

diff --git a/aws/resource_aws_dms_replication_instance_test.go b/aws/resource_aws_dms_replication_instance_test.go
index 3fc3e3922e9..71799f19b30 100644
--- a/aws/resource_aws_dms_replication_instance_test.go
+++ b/aws/resource_aws_dms_replication_instance_test.go
@@ -5,10 +5,10 @@ import (
 	"log"
 	"sort"
 	"testing"
-	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	dms "github.com/aws/aws-sdk-go/service/databasemigrationservice"
+	multierror "github.com/hashicorp/go-multierror"
 	gversion "github.com/hashicorp/go-version"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
@@ -24,57 +24,45 @@ func init() {
 
 func testSweepDmsReplicationInstances(region string) error {
 	client, err := sharedClientForRegion(region)
+
 	if err != nil {
 		return fmt.Errorf("error getting client: %s", err)
 	}
+
 	conn := client.(*AWSClient).dmsconn
+	sweepResources := make([]*testSweepResource, 0)
+	var errs *multierror.Error
 
 	err = conn.DescribeReplicationInstancesPages(&dms.DescribeReplicationInstancesInput{}, func(page *dms.DescribeReplicationInstancesOutput, lastPage bool) bool {
 		for _, instance := range page.ReplicationInstances {
-			arn := aws.StringValue(instance.ReplicationInstanceArn)
-			input := &dms.DeleteReplicationInstanceInput{
-				ReplicationInstanceArn: instance.ReplicationInstanceArn,
-			}
-
-			log.Printf("[INFO] Deleting DMS Replication Instance: %s", arn)
-			_, err := conn.DeleteReplicationInstance(input)
-
-			if isAWSErr(err, dms.ErrCodeResourceNotFoundFault, "") {
-				continue
-			}
-
-			if err != nil {
-				log.Printf("[ERROR] Error deleting DMS Replication Instance (%s): %s", arn, err)
-				continue
-			}
-
-			stateConf := &resource.StateChangeConf{
-				Pending:    []string{"deleting"},
-				Target:     []string{},
-				Refresh:    resourceAwsDmsReplicationInstanceStateRefreshFunc(conn, aws.StringValue(instance.ReplicationInstanceIdentifier)),
-				Timeout:    30 * time.Minute,
-				MinTimeout: 10 * time.Second,
-				Delay:      30 * time.Second,
-			}
+			r := resourceAwsDmsReplicationInstance()
+			d := r.Data(nil)
+			d.Set("replication_instance_arn", instance.ReplicationInstanceArn)
+			d.SetId(aws.StringValue(instance.ReplicationInstanceIdentifier))
 
-			if _, err := stateConf.WaitForState(); err != nil {
-				log.Printf("[ERROR] Error waiting for DMS Replication Instance (%s) deletion: %s", arn, err)
-			}
+			sweepResources = append(sweepResources, NewTestSweepResource(r, d, client))
 		}
 
 		return !lastPage
 	})
 
-	if testSweepSkipSweepError(err) {
-		log.Printf("[WARN] Skipping DMS Replication Instance sweep for %s: %s", region, err)
-		return nil
+	if err != nil {
+		errs = multierror.Append(errs, fmt.Errorf("error describing DMS Replication Instances: %w", err))
 	}
 
-	if err != nil {
-		return fmt.Errorf("Error retrieving DMS Replication Instances: %s", err)
+	if len(sweepResources) > 0 {
+		// any errors didn't prevent gathering of some work, so do it
+		if err := testSweepResourceOrchestrator(sweepResources); err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("error sweeping DMS Replication Instances for %s: %w", region, err))
+		}
 	}
 
-	return nil
+	if testSweepSkipSweepError(errs.ErrorOrNil()) {
+		log.Printf("[WARN] Skipping DMS Replication Instance sweep for %s: %s", region, err)
+		return nil
+	}
+
+	return errs.ErrorOrNil()
 }
 
 func TestAccAWSDmsReplicationInstance_basic(t *testing.T) {