DDB update replica Amazon Owned SSE

.changelog/31284.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+resource/aws_dynamodb_table: Fix changing replicas to the default `Managed by DynamoDB` encryption setting
+```

internal/service/dynamodb/table.go

@@ -1072,33 +1072,34 @@ func resourceTableUpdate(ctx context.Context, d *schema.ResourceData, meta inter
 	}
 
 	if d.HasChange("server_side_encryption") {
-		if replicas := d.Get("replica").(*schema.Set); replicas.Len() > 0 {
+		if replicas, sseSpecification := d.Get("replica").(*schema.Set), expandEncryptAtRestOptions(d.Get("server_side_encryption").([]interface{})); replicas.Len() > 0 && sseSpecification.KMSMasterKeyId != nil {
 			log.Printf("[DEBUG] Using SSE update on replicas")
 			var replicaInputs []awstypes.ReplicationGroupUpdate
-			var replicaRegions []string
 			for _, replica := range replicas.List() {
 				tfMap, ok := replica.(map[string]interface{})
 				if !ok {
 					continue
 				}
-				var regionName string
-				var KMSMasterKeyId string
-				if v, ok := tfMap["region_name"].(string); ok {
-					regionName = v
-					replicaRegions = append(replicaRegions, v)
+
+				region, ok := tfMap["region_name"].(string)
+				if !ok {
+					continue
 				}
-				if v, ok := tfMap[names.AttrKMSKeyARN].(string); ok && v != "" {
-					KMSMasterKeyId = v
+
+				key, ok := tfMap[names.AttrKMSKeyARN].(string)
+				if !ok || key == "" {
+					continue
 				}
+
 				var input = &awstypes.UpdateReplicationGroupMemberAction{
-					RegionName:     aws.String(regionName),
-					KMSMasterKeyId: aws.String(KMSMasterKeyId),
+					RegionName:     aws.String(region),
+					KMSMasterKeyId: aws.String(key),
 				}
 				var update = awstypes.ReplicationGroupUpdate{Update: input}
 				replicaInputs = append(replicaInputs, update)
 			}
 			var input = &awstypes.UpdateReplicationGroupMemberAction{
-				KMSMasterKeyId: expandEncryptAtRestOptions(d.Get("server_side_encryption").([]interface{})).KMSMasterKeyId,
+				KMSMasterKeyId: sseSpecification.KMSMasterKeyId,
 				RegionName:     aws.String(meta.(*conns.AWSClient).Region),
 			}
 			var update = awstypes.ReplicationGroupUpdate{Update: input}
@@ -1110,28 +1111,41 @@ func resourceTableUpdate(ctx context.Context, d *schema.ResourceData, meta inter
 			if err != nil {
 				return sdkdiag.AppendErrorf(diags, "updating DynamoDB Table (%s) SSE: %s", d.Id(), err)
 			}
-			for _, region := range replicaRegions {
-				if _, err := waitReplicaSSEUpdated(ctx, conn, region, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
-					return sdkdiag.AppendErrorf(diags, "waiting for DynamoDB Table (%s) replica SSE update in region %q: %s", d.Id(), region, err)
-				}
-			}
-			if _, err := waitSSEUpdated(ctx, conn, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
-				return sdkdiag.AppendErrorf(diags, "waiting for DynamoDB Table (%s) SSE update: %s", d.Id(), err)
-			}
 		} else {
 			log.Printf("[DEBUG] Using normal update for SSE")
 			_, err := conn.UpdateTable(ctx, &dynamodb.UpdateTableInput{
 				TableName:        aws.String(d.Id()),
-				SSESpecification: expandEncryptAtRestOptions(d.Get("server_side_encryption").([]interface{})),
+				SSESpecification: sseSpecification,
 			})
 			if err != nil {
 				return sdkdiag.AppendErrorf(diags, "updating DynamoDB Table (%s) SSE: %s", d.Id(), err)
 			}
+		}
 
-			if _, err := waitSSEUpdated(ctx, conn, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
-				return sdkdiag.AppendErrorf(diags, "waiting for DynamoDB Table (%s) SSE update: %s", d.Id(), err)
+		// since we don't update replicas unless there is a KMS key, we need to wait for replica
+		// updates for the scenario where 1) there are replicas, 2) we are updating SSE (such as
+		// disabling), and 3) we have no KMS key
+		if replicas := d.Get("replica").(*schema.Set); replicas.Len() > 0 {
+			var replicaRegions []string
+			for _, replica := range replicas.List() {
+				tfMap, ok := replica.(map[string]interface{})
+				if !ok {
+					continue
+				}
+				if v, ok := tfMap["region_name"].(string); ok {
+					replicaRegions = append(replicaRegions, v)
+				}
+			}
+			for _, region := range replicaRegions {
+				if _, err := waitReplicaSSEUpdated(ctx, conn, region, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
+					return sdkdiag.AppendErrorf(diags, "waiting for DynamoDB Table (%s) replica SSE update in region %q: %s", d.Id(), region, err)
+				}
 			}
 		}
+
+		if _, err := waitSSEUpdated(ctx, conn, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
+			return sdkdiag.AppendErrorf(diags, "waiting for DynamoDB Table (%s) SSE update: %s", d.Id(), err)
+		}
 	}
 
 	if d.HasChange("ttl") {
@@ -1902,7 +1916,7 @@ func clearSSEDefaultKey(ctx context.Context, client *conns.AWSClient, sseList []
 		return sseList
 	}
 
-	if sse[names.AttrKMSKeyARN].(string) == dk {
+	if v, ok := sse[names.AttrKMSKeyARN].(string); ok && v == dk {
 		sse[names.AttrKMSKeyARN] = ""
 		return []interface{}{sse}
 	}

internal/service/dynamodb/table_test.go

@@ -1882,29 +1882,68 @@ func TestAccDynamoDBTable_Replica_singleDefaultKeyEncrypted(t *testing.T) {
 		CheckDestroy:             testAccCheckTableDestroy(ctx),
 		Steps: []resource.TestStep{
 			{
-				Config: testAccTableConfig_replicaEncryptedDefault(rName),
+				Config: testAccTableConfig_replicaEncryptedDefault(rName, true),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					testAccCheckInitialTableExists(ctx, resourceName, &conf),
 					resource.TestCheckResourceAttr(resourceName, "replica.#", acctest.Ct1),
 					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.0.enabled", acctest.CtTrue),
 				),
 			},
 			{
-				Config: testAccTableConfig_replicaEncryptedDefault(rName),
+				Config: testAccTableConfig_replicaEncryptedDefault(rName, true),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					testAccCheckInitialTableExists(ctx, resourceName, &conf),
 					resource.TestCheckResourceAttr(resourceName, "replica.#", acctest.Ct1),
 					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.0.enabled", acctest.CtTrue),
 				),
 			},
 			{
-				Config:   testAccTableConfig_replicaEncryptedDefault(rName),
+				Config:   testAccTableConfig_replicaEncryptedDefault(rName, true),
 				PlanOnly: true,
 			},
 		},
 	})
 }
 
+func TestAccDynamoDBTable_Replica_singleDefaultKeyEncryptedAmazonOwned(t *testing.T) {
+	ctx := acctest.Context(t)
+	if testing.Short() {
+		t.Skip("skipping long-running test in short mode")
+	}
+
+	var conf awstypes.TableDescription
+	resourceName := "aws_dynamodb_table.test"
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck: func() {
+			acctest.PreCheck(ctx, t)
+			acctest.PreCheckMultipleRegion(t, 2)
+		},
+		ErrorCheck:               acctest.ErrorCheck(t, names.DynamoDBServiceID),
+		ProtoV5ProviderFactories: acctest.ProtoV5FactoriesMultipleRegions(ctx, t, 3), // 3 due to shared test configuration
+		CheckDestroy:             testAccCheckTableDestroy(ctx),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccTableConfig_replicaEncryptedDefault(rName, true),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckInitialTableExists(ctx, resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "replica.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.0.enabled", acctest.CtTrue),
+				),
+			},
+			{
+				Config: testAccTableConfig_replicaEncryptedDefault(rName, false),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					testAccCheckInitialTableExists(ctx, resourceName, &conf),
+					resource.TestCheckResourceAttr(resourceName, "replica.#", acctest.Ct1),
+					resource.TestCheckResourceAttr(resourceName, "server_side_encryption.#", acctest.Ct0),
+				),
+			},
+		},
+	})
+}
+
 func TestAccDynamoDBTable_Replica_singleCMK(t *testing.T) {
 	ctx := acctest.Context(t)
 	if testing.Short() {
@@ -3827,7 +3866,7 @@ resource "aws_dynamodb_table" "test" {
 `, rName))
 }
 
-func testAccTableConfig_replicaEncryptedDefault(rName string) string {
+func testAccTableConfig_replicaEncryptedDefault(rName string, sseEnabled bool) string {
 	return acctest.ConfigCompose(
 		acctest.ConfigMultipleRegionProvider(3), // Prevent "Provider configuration not present" errors
 		fmt.Sprintf(`
@@ -3848,14 +3887,14 @@ resource "aws_dynamodb_table" "test" {
   }
 
   server_side_encryption {
-    enabled = true
+    enabled = %[2]t
   }
 
   replica {
     region_name = data.aws_region.alternate.name
   }
 }
-`, rName))
+`, rName, sseEnabled))
 }
 
 func testAccTableConfig_replicaCMK(rName string) string {

internal/service/dynamodb/wait.go

@@ -204,13 +204,14 @@ func waitTTLUpdated(ctx context.Context, conn *dynamodb.Client, tableName string
 	return nil, err
 }
 
-func waitSSEUpdated(ctx context.Context, conn *dynamodb.Client, tableName string, timeout time.Duration) (*awstypes.TableDescription, error) { //nolint:unparam
+func waitSSEUpdated(ctx context.Context, conn *dynamodb.Client, tableName string, timeout time.Duration) (*awstypes.TableDescription, error) {
 	stateConf := &retry.StateChangeConf{
-		Pending: enum.Slice(awstypes.SSEStatusDisabling, awstypes.SSEStatusEnabling, awstypes.SSEStatusUpdating),
-		Target:  enum.Slice(awstypes.SSEStatusDisabled, awstypes.SSEStatusEnabled),
-		Refresh: statusSSE(ctx, conn, tableName),
-		Timeout: max(updateTableTimeout, timeout),
-		Delay:   30 * time.Second,
+		Pending:                   enum.Slice(awstypes.SSEStatusDisabling, awstypes.SSEStatusEnabling, awstypes.SSEStatusUpdating),
+		Target:                    enum.Slice(awstypes.SSEStatusDisabled, awstypes.SSEStatusEnabled),
+		Refresh:                   statusSSE(ctx, conn, tableName),
+		Timeout:                   max(updateTableTimeout, timeout),
+		ContinuousTargetOccurence: 2,
+		Delay:                     30 * time.Second,
 	}
 
 	outputRaw, err := stateConf.WaitForStateContext(ctx)
@@ -229,8 +230,9 @@ func waitReplicaSSEUpdated(ctx context.Context, conn *dynamodb.Client, region, t
 		Refresh: statusSSE(ctx, conn, tableName, func(o *dynamodb.Options) {
 			o.Region = region
 		}),
-		Timeout: max(updateTableTimeout, timeout),
-		Delay:   30 * time.Second,
+		Timeout:                   max(updateTableTimeout, timeout),
+		ContinuousTargetOccurence: 2,
+		Delay:                     30 * time.Second,
 	}
 
 	outputRaw, err := stateConf.WaitForStateContext(ctx)