From 6b3828b99784aae187aa42ea78de5234dc550183 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Wed, 10 May 2023 16:58:31 -0400
Subject: [PATCH 1/7] r/aws_msk_cluster: Remove
 'broker_node_group_info.ebs_volume_size'.

---
 .changelog/#####.txt                          |   3 +
 .../kafka/broker_nodes_data_source_test.go    |   9 +-
 internal/service/kafka/cluster.go             |  81 +----
 .../service/kafka/cluster_data_source_test.go |   9 +-
 internal/service/kafka/cluster_test.go        | 316 +++++++++---------
 .../kafka/scram_secret_association_test.go    |  17 +-
 .../service/kafkaconnect/connector_test.go    |   7 +-
 .../lambda/event_source_mapping_test.go       |  14 +-
 website/docs/r/msk_cluster.html.markdown      |   3 +-
 9 files changed, 222 insertions(+), 237 deletions(-)
 create mode 100644 .changelog/#####.txt

diff --git a/.changelog/#####.txt b/.changelog/#####.txt
new file mode 100644
index 000000000000..3d2a10dca127
--- /dev/null
+++ b/.changelog/#####.txt
@@ -0,0 +1,3 @@
+```release-note:note
+resource/aws_msk_cluster: The `broker_node_group_info.ebs_volume_size` attribute has been removed
+```
\ No newline at end of file
diff --git a/internal/service/kafka/broker_nodes_data_source_test.go b/internal/service/kafka/broker_nodes_data_source_test.go
index 72b0461d7e36..a46c936d793c 100644
--- a/internal/service/kafka/broker_nodes_data_source_test.go
+++ b/internal/service/kafka/broker_nodes_data_source_test.go
@@ -37,7 +37,7 @@ func TestAccKafkaBrokerNodesDataSource_basic(t *testing.T) {
 }
 
 func testAccBrokerNodesDataSourceConfig_basic(rName string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.2.1"
@@ -45,9 +45,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.t3.small"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   tags = {
diff --git a/internal/service/kafka/cluster.go b/internal/service/kafka/cluster.go
index 00273343309e..ebe78911af54 100644
--- a/internal/service/kafka/cluster.go
+++ b/internal/service/kafka/cluster.go
@@ -44,12 +44,6 @@ func ResourceCluster() *schema.Resource {
 			customdiff.ForceNewIfChange("kafka_version", func(_ context.Context, old, new, meta interface{}) bool {
 				return verify.SemVerLessThan(new.(string), old.(string))
 			}),
-			customdiff.ComputedIf("broker_node_group_info.0.storage_info", func(_ context.Context, diff *schema.ResourceDiff, meta interface{}) bool {
-				return diff.HasChange("broker_node_group_info.0.ebs_volume_size")
-			}),
-			customdiff.ComputedIf("broker_node_group_info.0.ebs_volume_size", func(_ context.Context, diff *schema.ResourceDiff, meta interface{}) bool {
-				return diff.HasChange("broker_node_group_info.0.storage_info")
-			}),
 			verify.SetTagsDiff,
 		),
 
@@ -134,14 +128,6 @@ func ResourceCluster() *schema.Resource {
 								},
 							},
 						},
-						"ebs_volume_size": {
-							Type:          schema.TypeInt,
-							Optional:      true,
-							Computed:      true,
-							Deprecated:    "use 'storage_info' argument instead",
-							ValidateFunc:  validation.IntBetween(1, 16384),
-							ConflictsWith: []string{"broker_node_group_info.0.storage_info"},
-						},
 						"instance_type": {
 							Type:     schema.TypeString,
 							Required: true,
@@ -155,11 +141,10 @@ func ResourceCluster() *schema.Resource {
 							},
 						},
 						"storage_info": {
-							Type:          schema.TypeList,
-							Optional:      true,
-							Computed:      true,
-							MaxItems:      1,
-							ConflictsWith: []string{"broker_node_group_info.0.ebs_volume_size"},
+							Type:     schema.TypeList,
+							Optional: true,
+							Computed: true,
+							MaxItems: 1,
 							Elem: &schema.Resource{
 								Schema: map[string]*schema.Schema{
 									"ebs_storage_info": {
@@ -671,52 +656,32 @@ func resourceClusterUpdate(ctx context.Context, d *schema.ResourceData, meta int
 		}
 	}
 
-	if d.HasChanges("broker_node_group_info.0.ebs_volume_size", "broker_node_group_info.0.storage_info") {
+	if d.HasChanges("broker_node_group_info.0.storage_info") {
 		input := &kafka.UpdateBrokerStorageInput{
 			ClusterArn:     aws.String(d.Id()),
 			CurrentVersion: aws.String(d.Get("current_version").(string)),
-		}
-		if d.HasChange("broker_node_group_info.0.storage_info") {
-			// case 1: deprecated ebs_volume_size replaced with storage_info
-			// case 2: regular update of storage_info
-			ebsVolumeInfo := &kafka.BrokerEBSVolumeInfo{
+			TargetBrokerEBSVolumeInfo: []*kafka.BrokerEBSVolumeInfo{{
 				KafkaBrokerNodeId: aws.String("All"),
 				VolumeSizeGB:      aws.Int64(int64(d.Get("broker_node_group_info.0.storage_info.0.ebs_storage_info.0.volume_size").(int))),
-			}
-			if v, ok := d.GetOk("broker_node_group_info.0.storage_info.0.ebs_storage_info.0.provisioned_throughput"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
-				ebsVolumeInfo.ProvisionedThroughput = expandProvisionedThroughput(v.([]interface{})[0].(map[string]interface{}))
-			}
-			input.TargetBrokerEBSVolumeInfo = []*kafka.BrokerEBSVolumeInfo{ebsVolumeInfo}
-		} else {
-			// case 3: regular update of deprecated ebs_volume_size
-			input.TargetBrokerEBSVolumeInfo = []*kafka.BrokerEBSVolumeInfo{
-				{
-					KafkaBrokerNodeId: aws.String("All"),
-					VolumeSizeGB:      aws.Int64(int64(d.Get("broker_node_group_info.0.ebs_volume_size").(int))),
-				},
-			}
+			}},
+		}
+
+		if v, ok := d.GetOk("broker_node_group_info.0.storage_info.0.ebs_storage_info.0.provisioned_throughput"); ok && len(v.([]interface{})) > 0 && v.([]interface{})[0] != nil {
+			input.TargetBrokerEBSVolumeInfo[0].ProvisionedThroughput = expandProvisionedThroughput(v.([]interface{})[0].(map[string]interface{}))
 		}
 
 		output, err := conn.UpdateBrokerStorageWithContext(ctx, input)
 
-		// the following error is thrown if previous ebs_volume_size and new storage_info.ebs_storage_info.volume_size have the same value:
-		// BadRequestException: The request does not include any updates to the EBS volumes of the cluster. Verify the request, then try again
-		// ignore this error to allow users to replace deprecated ebs_volume_size with storage_info - Address case 1
-		if err != nil && !tfawserr.ErrMessageContains(err, kafka.ErrCodeBadRequestException, "The request does not include any updates to the EBS volumes of the cluster") {
+		if err != nil {
 			return diag.Errorf("updating MSK Cluster (%s) broker storage: %s", d.Id(), err)
 		}
 
 		clusterOperationARN := aws.StringValue(output.ClusterOperationArn)
 
-		// when there are no changes, output.ClusterOperationArn is not returned leading to
-		// InvalidParameter: 1 validation error(s) found. - minimum field size of 1, DescribeClusterOperationInput.ClusterOperationArn.
-		// skip the wait if the EBS volume size is unchanged
-		if !tfawserr.ErrMessageContains(err, kafka.ErrCodeBadRequestException, "The request does not include any updates to the EBS volumes of the cluster") {
-			_, err = waitClusterOperationCompleted(ctx, conn, clusterOperationARN, d.Timeout(schema.TimeoutUpdate))
+		_, err = waitClusterOperationCompleted(ctx, conn, clusterOperationARN, d.Timeout(schema.TimeoutUpdate))
 
-			if err != nil {
-				return diag.Errorf("waiting for MSK Cluster (%s) operation (%s): %s", d.Id(), clusterOperationARN, err)
-			}
+		if err != nil {
+			return diag.Errorf("waiting for MSK Cluster (%s) operation (%s): %s", d.Id(), clusterOperationARN, err)
 		}
 
 		// refresh the current_version attribute after each update
@@ -979,14 +944,6 @@ func expandBrokerNodeGroupInfo(tfMap map[string]interface{}) *kafka.BrokerNodeGr
 		apiObject.SecurityGroups = flex.ExpandStringSet(v)
 	}
 
-	if v, ok := tfMap["ebs_volume_size"].(int); ok && v != 0 {
-		apiObject.StorageInfo = &kafka.StorageInfo{
-			EbsStorageInfo: &kafka.EBSStorageInfo{
-				VolumeSize: aws.Int64(int64(v)),
-			},
-		}
-	}
-
 	if v, ok := tfMap["storage_info"].([]interface{}); ok && len(v) > 0 && v[0] != nil {
 		apiObject.StorageInfo = expandStorageInfo(v[0].(map[string]interface{}))
 	}
@@ -1376,14 +1333,6 @@ func flattenBrokerNodeGroupInfo(apiObject *kafka.BrokerNodeGroupInfo) map[string
 		tfMap["storage_info"] = flattenStorageInfo(v)
 	}
 
-	if v := apiObject.StorageInfo; v != nil {
-		if v := v.EbsStorageInfo; v != nil {
-			if v := v.VolumeSize; v != nil {
-				tfMap["ebs_volume_size"] = aws.Int64Value(v)
-			}
-		}
-	}
-
 	return tfMap
 }
 
diff --git a/internal/service/kafka/cluster_data_source_test.go b/internal/service/kafka/cluster_data_source_test.go
index 360d38987a93..fd3bd7deaa71 100644
--- a/internal/service/kafka/cluster_data_source_test.go
+++ b/internal/service/kafka/cluster_data_source_test.go
@@ -46,7 +46,7 @@ func TestAccKafkaClusterDataSource_basic(t *testing.T) {
 }
 
 func testAccClusterDataSourceConfig_basic(rName string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.2.1"
@@ -54,9 +54,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   tags = {
diff --git a/internal/service/kafka/cluster_test.go b/internal/service/kafka/cluster_test.go
index 86ad6047d01a..471f7bd8cc0c 100644
--- a/internal/service/kafka/cluster_test.go
+++ b/internal/service/kafka/cluster_test.go
@@ -71,7 +71,6 @@ func TestAccKafkaCluster_basic(t *testing.T) {
 					testAccCheckResourceAttrIsSortedCSV(resourceName, "bootstrap_brokers_tls"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.az_distribution", kafka.BrokerAZDistributionDefault),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.ebs_volume_size", "10"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.client_subnets.#", "3"),
 					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az1", "id"),
 					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az2", "id"),
@@ -82,6 +81,11 @@ func TestAccKafkaCluster_basic(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.instance_type", "kafka.m5.large"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.security_groups.#", "1"),
 					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.security_groups.*", "aws_security_group.example_sg", "id"),
+					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.volume_size", "10"),
+					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.provisioned_throughput.#", "0"),
 					resource.TestCheckResourceAttr(resourceName, "client_authentication.#", "0"),
 					resource.TestCheckResourceAttr(resourceName, "cluster_name", rName),
 					resource.TestCheckResourceAttr(resourceName, "configuration_info.#", "1"),
@@ -185,48 +189,6 @@ func TestAccKafkaCluster_tags(t *testing.T) {
 	})
 }
 
-func TestAccKafkaCluster_BrokerNodeGroupInfo_ebsVolumeSize(t *testing.T) {
-	ctx := acctest.Context(t)
-	var cluster1, cluster2 kafka.ClusterInfo
-	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
-	resourceName := "aws_msk_cluster.test"
-
-	resource.ParallelTest(t, resource.TestCase{
-		PreCheck:                 func() { acctest.PreCheck(ctx, t); testAccPreCheck(ctx, t) },
-		ErrorCheck:               acctest.ErrorCheck(t, kafka.EndpointsID),
-		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
-		CheckDestroy:             testAccCheckClusterDestroy(ctx),
-		Steps: []resource.TestStep{
-			{
-				Config: testAccClusterConfig_deprecatedBrokerNodeGroupInfoEBSVolumeSize(rName, 11),
-				Check: resource.ComposeAggregateTestCheckFunc(
-					testAccCheckClusterExists(ctx, resourceName, &cluster1),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.ebs_volume_size", "11"),
-				),
-			},
-			{
-				ResourceName:      resourceName,
-				ImportState:       true,
-				ImportStateVerify: true,
-				ImportStateVerifyIgnore: []string{
-					"current_version",
-				},
-			},
-			{
-				// BadRequestException: The minimum increase in storage size of the cluster should be atleast 100GB
-				Config: testAccClusterConfig_deprecatedBrokerNodeGroupInfoEBSVolumeSize(rName, 112),
-				Check: resource.ComposeAggregateTestCheckFunc(
-					testAccCheckClusterExists(ctx, resourceName, &cluster2),
-					testAccCheckClusterNotRecreated(&cluster1, &cluster2),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.ebs_volume_size", "112"),
-				),
-			},
-		},
-	})
-}
-
 func TestAccKafkaCluster_BrokerNodeGroupInfo_storageInfo(t *testing.T) {
 	ctx := acctest.Context(t)
 	var cluster1, cluster2 kafka.ClusterInfo
@@ -246,7 +208,6 @@ func TestAccKafkaCluster_BrokerNodeGroupInfo_storageInfo(t *testing.T) {
 				Check: resource.ComposeAggregateTestCheckFunc(
 					testAccCheckClusterExists(ctx, resourceName, &cluster1),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.ebs_volume_size", strconv.Itoa(original_volume_size)),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.volume_size", strconv.Itoa(original_volume_size)),
@@ -265,54 +226,6 @@ func TestAccKafkaCluster_BrokerNodeGroupInfo_storageInfo(t *testing.T) {
 			{
 				// update broker_node_group_info.0.storage_info.0.ebs_storage_info.0.volume_size
 				Config: testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeSetAndProvThroughputEnabled(rName, updated_volume_size, "kafka.m5.4xlarge"),
-				Check: resource.ComposeAggregateTestCheckFunc(
-					testAccCheckClusterExists(ctx, resourceName, &cluster2),
-					testAccCheckClusterNotRecreated(&cluster1, &cluster2),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.ebs_volume_size", strconv.Itoa(updated_volume_size)),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.volume_size", strconv.Itoa(updated_volume_size)),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.provisioned_throughput.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.provisioned_throughput.0.enabled", "true"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.0.provisioned_throughput.0.volume_throughput", "250"),
-				),
-			},
-		},
-	})
-}
-
-func TestAccKafkaCluster_BrokerNodeGroupInfo_modifyEBSVolumeSizeToStorageInfo(t *testing.T) {
-	ctx := acctest.Context(t)
-	var cluster1, cluster2 kafka.ClusterInfo
-	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
-	resourceName := "aws_msk_cluster.test"
-	original_volume_size := 11
-	updated_volume_size := 112
-
-	resource.ParallelTest(t, resource.TestCase{
-		PreCheck:                 func() { acctest.PreCheck(ctx, t); testAccPreCheck(ctx, t) },
-		ErrorCheck:               acctest.ErrorCheck(t, kafka.EndpointsID),
-		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
-		CheckDestroy:             testAccCheckClusterDestroy(ctx),
-		Steps: []resource.TestStep{
-			{
-				// init with the deprecated ebs_volume_size
-				Config: testAccClusterConfig_deprecatedBrokerNodeGroupInfoEBSVolumeSize(rName, original_volume_size),
-				Check: resource.ComposeAggregateTestCheckFunc(
-					testAccCheckClusterExists(ctx, resourceName, &cluster1),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
-					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.ebs_volume_size", strconv.Itoa(original_volume_size)),
-				),
-			},
-			{
-				// refactor deprecated ebs_volume_size to storage_info
-				Config:   testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeOnly(rName, original_volume_size, "kafka.m5.large"),
-				PlanOnly: true,
-			},
-			{
-				// upgrade the instance type, update storage, and enable provisioned throughput
-				Config: testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeSetAndProvThroughputEnabled(rName, updated_volume_size, "kafka.m5.4xlarge"),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					testAccCheckClusterExists(ctx, resourceName, &cluster2),
 					testAccCheckClusterNotRecreated(&cluster1, &cluster2),
@@ -1317,7 +1230,7 @@ func testAccPreCheck(ctx context.Context, t *testing.T) {
 	}
 }
 
-func testAccClusterBaseConfig(rName string) string {
+func testAccClusterConfig_base(rName string) string {
 	return acctest.ConfigCompose(acctest.ConfigAvailableAZsNoOptIn(), fmt.Sprintf(`
 resource "aws_vpc" "example_vpc" {
   cidr_block = "192.168.0.0/22"
@@ -1487,7 +1400,7 @@ resource "aws_route_table_association" "route_tbl_assoc_3" {
 }
 
 func testAccClusterConfig_basic(rName string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1495,33 +1408,21 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
-  }
-}
-`, rName))
-}
 
-func testAccClusterConfig_deprecatedBrokerNodeGroupInfoEBSVolumeSize(rName string, ebsVolumeSize int) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
-resource "aws_msk_cluster" "test" {
-  cluster_name           = %[1]q
-  kafka_version          = "2.7.1"
-  number_of_broker_nodes = 3
-
-  broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = %[2]d
-    instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
-`, rName, ebsVolumeSize))
+`, rName))
 }
 
 func testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeOnly(rName string, ebsVolumeSize int, instanceType string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1542,7 +1443,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeSetAndProvThroughputNotEnabled(rName string, ebsVolumeSize int, instanceType string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1566,7 +1467,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeSetAndProvThroughputEnabled(rName string, ebsVolumeSize int, instanceType string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1591,7 +1492,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_brokerNodeGroupInfoInstanceType(rName string, t string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1599,9 +1500,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = %[2]q
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 `, rName, t))
@@ -1631,9 +1537,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_public_subnet_az1.id, aws_subnet.example_public_subnet_az2.id, aws_subnet.example_public_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   configuration_info {
@@ -1662,10 +1573,15 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_public_subnet_az1.id, aws_subnet.example_public_subnet_az2.id, aws_subnet.example_public_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
 
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
+
     connectivity_info {
       public_access {
         type = %[2]q
@@ -1707,7 +1623,7 @@ resource "aws_acmpca_certificate_authority" "test" {
 
 func testAccClusterConfig_clientAuthenticationTLSCertificateAuthorityARNs(rName, commonName string) string {
 	return acctest.ConfigCompose(
-		testAccClusterBaseConfig(rName),
+		testAccClusterConfig_base(rName),
 		testAccClusterConfig_rootCA(commonName),
 		fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
@@ -1717,9 +1633,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   client_authentication {
@@ -1744,7 +1665,7 @@ resource "aws_msk_cluster" "test" {
 
 func testAccClusterConfig_rootCANoClientAuthentication(rName, commonName string) string {
 	return acctest.ConfigCompose(
-		testAccClusterBaseConfig(rName),
+		testAccClusterConfig_base(rName),
 		testAccClusterConfig_rootCA(commonName),
 		fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
@@ -1754,9 +1675,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   encryption_info {
@@ -1770,7 +1696,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_clientAuthenticationSASLScram(rName string, scramEnabled bool) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1778,9 +1704,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   client_authentication {
@@ -1795,7 +1726,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_clientAuthenticationSASLIAM(rName string, saslEnabled bool) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1803,9 +1734,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   client_authentication {
@@ -1820,7 +1756,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_configurationInfoRevision1(rName string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_configuration" "test1" {
   kafka_versions = ["2.7.1"]
   name           = "%[1]s-1"
@@ -1837,9 +1773,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   configuration_info {
@@ -1851,7 +1792,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_configurationInfoRevision2(rName string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_configuration" "test1" {
   kafka_versions = ["2.7.1"]
   name           = "%[1]s-1"
@@ -1877,9 +1818,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   configuration_info {
@@ -1891,7 +1837,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_encryptionInfoEncryptionAtRestKMSKeyARN(rName string) string { // nosemgrep:ci.msk-in-func-name
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_kms_key" "example_key" {
   description = %[1]q
 
@@ -1907,9 +1853,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   encryption_info {
@@ -1920,7 +1871,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_encryptionInfoEncryptionInTransitClientBroker(rName, clientBroker string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1928,9 +1879,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   encryption_info {
@@ -1943,7 +1899,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_encryptionInfoEncryptionInTransitIn(rName string, inCluster bool) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -1951,9 +1907,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   encryption_info {
@@ -1966,7 +1927,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_enhancedMonitoring(rName, enhancedMonitoring string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   enhanced_monitoring    = %[2]q
@@ -1975,16 +1936,21 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 `, rName, enhancedMonitoring))
 }
 
 func testAccClusterConfig_storageMode(rName string, storageMode string, kafkaVersion string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   storage_mode           = %[2]q
@@ -1993,16 +1959,21 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 `, rName, storageMode, kafkaVersion))
 }
 
 func testAccClusterConfig_numberOfBrokerNodes(rName string, brokerCount int) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -2010,16 +1981,21 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 `, rName, brokerCount))
 }
 
 func testAccClusterConfig_openMonitoring(rName string, jmxExporterEnabled bool, nodeExporterEnabled bool) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -2027,9 +2003,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   open_monitoring {
@@ -2062,7 +2043,7 @@ func testAccClusterConfig_loggingInfo(rName string, cloudwatchLogsEnabled bool,
 		s3Bucket = "aws_s3_bucket.bucket.id"
 	}
 
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_cloudwatch_log_group" "test" {
   name = %[1]q
 }
@@ -2125,9 +2106,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   logging_info {
@@ -2154,7 +2140,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_version(rName string, kafkaVersion string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = %[2]q
@@ -2168,16 +2154,21 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 `, rName, kafkaVersion))
 }
 
 func testAccClusterConfig_versionConfigurationInfo(rName string, kafkaVersion string, configResourceName string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_configuration" "config1" {
   kafka_versions    = ["2.7.1"]
   name              = "%[1]s-1"
@@ -2207,9 +2198,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   configuration_info {
@@ -2221,7 +2217,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_tags1(rName, tagKey1, tagValue1 string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -2229,9 +2225,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   tags = {
@@ -2242,7 +2243,7 @@ resource "aws_msk_cluster" "test" {
 }
 
 func testAccClusterConfig_tags2(rName, tagKey1, tagValue1, tagKey2, tagValue2 string) string {
-	return acctest.ConfigCompose(testAccClusterBaseConfig(rName), fmt.Sprintf(`
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
   kafka_version          = "2.7.1"
@@ -2250,9 +2251,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   tags = {
diff --git a/internal/service/kafka/scram_secret_association_test.go b/internal/service/kafka/scram_secret_association_test.go
index 8a27986e7e60..c8e30c212274 100644
--- a/internal/service/kafka/scram_secret_association_test.go
+++ b/internal/service/kafka/scram_secret_association_test.go
@@ -184,8 +184,8 @@ func testAccCheckScramSecretAssociationExists(ctx context.Context, resourceName
 	}
 }
 
-func testAccScramSecretAssociationBaseConfig(rName string, count int) string {
-	return fmt.Sprintf(`
+func testAccScramSecretAssociationConfig_base(rName string, count int) string {
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 data "aws_partition" "current" {}
 
 resource "aws_msk_cluster" "test" {
@@ -195,9 +195,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.t3.small"
     security_groups = [aws_security_group.example_sg.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 
   client_authentication {
@@ -242,13 +247,11 @@ resource "aws_secretsmanager_secret_policy" "test" {
 }
 POLICY
 }
-`, rName, count)
+`, rName, count))
 }
 
 func testAccScramSecretAssociationConfig_basic(rName string, count int) string {
-	return acctest.ConfigCompose(
-		testAccClusterBaseConfig(rName),
-		testAccScramSecretAssociationBaseConfig(rName, count), `
+	return acctest.ConfigCompose(testAccScramSecretAssociationConfig_base(rName, count), `
 resource "aws_msk_scram_secret_association" "test" {
   cluster_arn     = aws_msk_cluster.test.arn
   secret_arn_list = aws_secretsmanager_secret.test[*].arn
diff --git a/internal/service/kafkaconnect/connector_test.go b/internal/service/kafkaconnect/connector_test.go
index 9501b57ffbfa..2641da8be187 100644
--- a/internal/service/kafkaconnect/connector_test.go
+++ b/internal/service/kafkaconnect/connector_test.go
@@ -397,9 +397,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = [aws_subnet.test1.id, aws_subnet.test2.id, aws_subnet.test3.id]
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.test.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 `, rName))
diff --git a/internal/service/lambda/event_source_mapping_test.go b/internal/service/lambda/event_source_mapping_test.go
index f0c32bcff7db..3237266e6891 100644
--- a/internal/service/lambda/event_source_mapping_test.go
+++ b/internal/service/lambda/event_source_mapping_test.go
@@ -2199,9 +2199,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = aws_subnet.test[*].id
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.test.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 
@@ -2231,9 +2236,14 @@ resource "aws_msk_cluster" "test" {
 
   broker_node_group_info {
     client_subnets  = aws_subnet.test[*].id
-    ebs_volume_size = 10
     instance_type   = "kafka.m5.large"
     security_groups = [aws_security_group.test.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
   }
 }
 
diff --git a/website/docs/r/msk_cluster.html.markdown b/website/docs/r/msk_cluster.html.markdown
index 3105584b7157..85a4190d4d15 100644
--- a/website/docs/r/msk_cluster.html.markdown
+++ b/website/docs/r/msk_cluster.html.markdown
@@ -218,7 +218,6 @@ The following arguments are supported:
 ### broker_node_group_info Argument Reference
 
 * `client_subnets` - (Required) A list of subnets to connect to in client VPC ([documentation](https://docs.aws.amazon.com/msk/1.0/apireference/clusters.html#clusters-prop-brokernodegroupinfo-clientsubnets)).
-* `ebs_volume_size` - (Optional, **Deprecated** use `storage_info.ebs_storage_info.volume_size` instead) The size in GiB of the EBS volume for the data drive on each broker node.
 * `instance_type` - (Required) Specify the instance type to use for the kafka brokersE.g., kafka.m5.large. ([Pricing info](https://aws.amazon.com/msk/pricing/))
 * `security_groups` - (Required) A list of the security groups to associate with the elastic network interfaces to control who can communicate with the cluster.
 * `az_distribution` - (Optional) The distribution of broker nodes across availability zones ([documentation](https://docs.aws.amazon.com/msk/1.0/apireference/clusters.html#clusters-model-brokerazdistribution)). Currently the only valid value is `DEFAULT`.
@@ -339,7 +338,7 @@ In addition to all arguments above, the following attributes are exported:
 
 * `create` - (Default `120m`)
 * `update` - (Default `120m`)
-Note that the `update` timeout is used separately for `ebs_volume_size`, `instance_type`, `number_of_broker_nodes`, `configuration_info`, `kafka_version` and monitoring and logging update timeouts.
+Note that the `update` timeout is used separately for `storage_info`, `instance_type`, `number_of_broker_nodes`, `configuration_info`, `kafka_version` and monitoring and logging update timeouts.
 * `delete` - (Default `120m`)
 
 ## Import

From 9b5c3e93a8c5e2638dba962c3b71b8c356a18fa2 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Wed, 10 May 2023 17:02:25 -0400
Subject: [PATCH 2/7] Correct CHANGELOG entry file name.

---
 .changelog/{#####.txt => 31324.txt} | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename .changelog/{#####.txt => 31324.txt} (100%)

diff --git a/.changelog/#####.txt b/.changelog/31324.txt
similarity index 100%
rename from .changelog/#####.txt
rename to .changelog/31324.txt

From 3c983d2ce9b1dd3e470ee7242e974a9c6c0ef594 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Wed, 10 May 2023 17:21:45 -0400
Subject: [PATCH 3/7] Kafka: Tidy up acceptance test configurations.

---
 .../kafka/broker_nodes_data_source_test.go    |   4 +-
 .../service/kafka/cluster_data_source_test.go |   4 +-
 internal/service/kafka/cluster_test.go        | 272 ++++++------------
 .../kafka/scram_secret_association_test.go    |   4 +-
 4 files changed, 95 insertions(+), 189 deletions(-)

diff --git a/internal/service/kafka/broker_nodes_data_source_test.go b/internal/service/kafka/broker_nodes_data_source_test.go
index a46c936d793c..0899333e850e 100644
--- a/internal/service/kafka/broker_nodes_data_source_test.go
+++ b/internal/service/kafka/broker_nodes_data_source_test.go
@@ -44,9 +44,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.t3.small"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
diff --git a/internal/service/kafka/cluster_data_source_test.go b/internal/service/kafka/cluster_data_source_test.go
index fd3bd7deaa71..455d86d8dfda 100644
--- a/internal/service/kafka/cluster_data_source_test.go
+++ b/internal/service/kafka/cluster_data_source_test.go
@@ -53,9 +53,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
diff --git a/internal/service/kafka/cluster_test.go b/internal/service/kafka/cluster_test.go
index 471f7bd8cc0c..b2327bce1be2 100644
--- a/internal/service/kafka/cluster_test.go
+++ b/internal/service/kafka/cluster_test.go
@@ -72,15 +72,15 @@ func TestAccKafkaCluster_basic(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.az_distribution", kafka.BrokerAZDistributionDefault),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.client_subnets.#", "3"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az1", "id"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az2", "id"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az3", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.0", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.1", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.2", "id"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.connectivity_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.connectivity_info.0.public_access.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.connectivity_info.0.public_access.0.type", "DISABLED"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.instance_type", "kafka.m5.large"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.security_groups.#", "1"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.security_groups.*", "aws_security_group.example_sg", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.security_groups.*", "aws_security_group.test", "id"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.storage_info.0.ebs_storage_info.#", "1"),
@@ -825,9 +825,9 @@ func TestAccKafkaCluster_numberOfBrokerNodes(t *testing.T) {
 					resource.TestMatchResourceAttr(resourceName, "bootstrap_brokers_tls", clusterBoostrapBrokersTLSRegexp),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.client_subnets.#", "3"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az1", "id"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az2", "id"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az3", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.0", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.1", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.2", "id"),
 					resource.TestCheckResourceAttr(resourceName, "number_of_broker_nodes", "3"),
 					testAccCheckResourceAttrIsSortedCSV(resourceName, "bootstrap_brokers_tls"),
 				),
@@ -850,9 +850,9 @@ func TestAccKafkaCluster_numberOfBrokerNodes(t *testing.T) {
 					resource.TestMatchResourceAttr(resourceName, "bootstrap_brokers_tls", clusterBoostrapBrokersTLSRegexp),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.client_subnets.#", "3"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az1", "id"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az2", "id"),
-					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.example_subnet_az3", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.0", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.1", "id"),
+					resource.TestCheckTypeSetElemAttrPair(resourceName, "broker_node_group_info.0.client_subnets.*", "aws_subnet.test.2", "id"),
 					resource.TestCheckResourceAttr(resourceName, "number_of_broker_nodes", "6"),
 					testAccCheckResourceAttrIsSortedCSV(resourceName, "bootstrap_brokers_tls"),
 				),
@@ -1231,47 +1231,10 @@ func testAccPreCheck(ctx context.Context, t *testing.T) {
 }
 
 func testAccClusterConfig_base(rName string) string {
-	return acctest.ConfigCompose(acctest.ConfigAvailableAZsNoOptIn(), fmt.Sprintf(`
-resource "aws_vpc" "example_vpc" {
-  cidr_block = "192.168.0.0/22"
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_subnet" "example_subnet_az1" {
-  vpc_id            = aws_vpc.example_vpc.id
-  cidr_block        = "192.168.0.0/24"
-  availability_zone = data.aws_availability_zones.available.names[0]
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_subnet" "example_subnet_az2" {
-  vpc_id            = aws_vpc.example_vpc.id
-  cidr_block        = "192.168.1.0/24"
-  availability_zone = data.aws_availability_zones.available.names[1]
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_subnet" "example_subnet_az3" {
-  vpc_id            = aws_vpc.example_vpc.id
-  cidr_block        = "192.168.2.0/24"
-  availability_zone = data.aws_availability_zones.available.names[2]
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_security_group" "example_sg" {
-  vpc_id = aws_vpc.example_vpc.id
+	return acctest.ConfigCompose(acctest.ConfigVPCWithSubnets(rName, 3), fmt.Sprintf(`
+resource "aws_security_group" "test" {
+  name   = %[1]q
+  vpc_id = aws_vpc.test.id
 
   tags = {
     Name = %[1]q
@@ -1280,69 +1243,22 @@ resource "aws_security_group" "example_sg" {
 `, rName))
 }
 
-func testAccClusterBasePublicAccessConfig(rName string) string {
+func testAccClusterConfig_basePublicAccess(rName string) string {
 	return acctest.ConfigCompose(acctest.ConfigAvailableAZsNoOptIn(), fmt.Sprintf(`
-resource "aws_vpc" "example_vpc" {
-  cidr_block = "192.168.0.0/21"
+resource "aws_vpc" "test" {
+  cidr_block = "10.0.0.0/16"
 
   tags = {
     Name = %[1]q
   }
 }
 
-resource "aws_subnet" "example_subnet_az1" {
-  vpc_id            = aws_vpc.example_vpc.id
-  cidr_block        = "192.168.0.0/24"
-  availability_zone = data.aws_availability_zones.available.names[0]
-
-  tags = {
-    Name = %[1]q
-  }
-}
+resource "aws_subnet" "test" {
+  count = 3
 
-resource "aws_subnet" "example_subnet_az2" {
-  vpc_id            = aws_vpc.example_vpc.id
-  cidr_block        = "192.168.1.0/24"
-  availability_zone = data.aws_availability_zones.available.names[1]
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_subnet" "example_subnet_az3" {
-  vpc_id            = aws_vpc.example_vpc.id
-  cidr_block        = "192.168.2.0/24"
-  availability_zone = data.aws_availability_zones.available.names[2]
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_security_group" "example_sg" {
-  vpc_id = aws_vpc.example_vpc.id
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_subnet" "example_public_subnet_az1" {
-  vpc_id                  = aws_vpc.example_vpc.id
-  cidr_block              = "192.168.3.0/24"
-  availability_zone       = data.aws_availability_zones.available.names[0]
-  map_public_ip_on_launch = true
-
-  tags = {
-    Name = %[1]q
-  }
-}
-
-resource "aws_subnet" "example_public_subnet_az2" {
-  vpc_id                  = aws_vpc.example_vpc.id
-  cidr_block              = "192.168.4.0/24"
-  availability_zone       = data.aws_availability_zones.available.names[1]
+  vpc_id                  = aws_vpc.test.id
+  availability_zone       = data.aws_availability_zones.available.names[count.index]
+  cidr_block              = cidrsubnet(aws_vpc.test.cidr_block, 8, count.index)
   map_public_ip_on_launch = true
 
   tags = {
@@ -1350,31 +1266,29 @@ resource "aws_subnet" "example_public_subnet_az2" {
   }
 }
 
-resource "aws_subnet" "example_public_subnet_az3" {
-  vpc_id                  = aws_vpc.example_vpc.id
-  cidr_block              = "192.168.5.0/24"
-  availability_zone       = data.aws_availability_zones.available.names[2]
-  map_public_ip_on_launch = true
+resource "aws_security_group" "test" {
+  name   = %[1]q
+  vpc_id = aws_vpc.test.id
 
   tags = {
     Name = %[1]q
   }
 }
 
-resource "aws_internet_gateway" "example_igw" {
-  vpc_id = aws_vpc.example_vpc.id
+resource "aws_internet_gateway" "test" {
+  vpc_id = aws_vpc.test.id
 
   tags = {
     Name = %[1]q
   }
 }
 
-resource "aws_route_table" "example_route_tbl" {
-  vpc_id = aws_vpc.example_vpc.id
+resource "aws_route_table" "test" {
+  vpc_id = aws_vpc.test.id
 
   route {
     cidr_block = "0.0.0.0/0"
-    gateway_id = aws_internet_gateway.example_igw.id
+    gateway_id = aws_internet_gateway.test.id
   }
 
   tags = {
@@ -1382,19 +1296,11 @@ resource "aws_route_table" "example_route_tbl" {
   }
 }
 
-resource "aws_route_table_association" "route_tbl_assoc_1" {
-  subnet_id      = aws_subnet.example_public_subnet_az1.id
-  route_table_id = aws_route_table.example_route_tbl.id
-}
-
-resource "aws_route_table_association" "route_tbl_assoc_2" {
-  subnet_id      = aws_subnet.example_public_subnet_az2.id
-  route_table_id = aws_route_table.example_route_tbl.id
-}
+resource "aws_route_table_association" "test" {
+  count = 3
 
-resource "aws_route_table_association" "route_tbl_assoc_3" {
-  subnet_id      = aws_subnet.example_public_subnet_az3.id
-  route_table_id = aws_route_table.example_route_tbl.id
+  subnet_id      = aws_subnet.test[count.index].id
+  route_table_id = aws_route_table.test.id
 }
 `, rName))
 }
@@ -1407,9 +1313,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1429,9 +1335,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = %[3]q
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
     storage_info {
       ebs_storage_info {
         volume_size = %[2]d
@@ -1450,9 +1356,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = %[3]q
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
     storage_info {
       ebs_storage_info {
         provisioned_throughput {
@@ -1474,9 +1380,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = %[3]q
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
     storage_info {
       ebs_storage_info {
         provisioned_throughput {
@@ -1499,9 +1405,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = %[2]q
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1513,7 +1419,7 @@ resource "aws_msk_cluster" "test" {
 `, rName, t))
 }
 
-func testAccConfigurationAllowEveryoneNoACLFoundFalse(rName string) string {
+func testAccClusterConfig_allowEveryoneNoACLFoundFalse(rName string) string {
 	return fmt.Sprintf(`
 resource "aws_msk_configuration" "test" {
   kafka_versions = ["2.7.1"]
@@ -1527,8 +1433,8 @@ resource "aws_msk_configuration" "test" {
 
 func testAccClusterConfig_brokerNodeGroupInfoNoPublicAccessSASLIAM(rName string) string {
 	return acctest.ConfigCompose(
-		testAccClusterBasePublicAccessConfig(rName),
-		testAccConfigurationAllowEveryoneNoACLFoundFalse(rName),
+		testAccClusterConfig_basePublicAccess(rName),
+		testAccClusterConfig_allowEveryoneNoACLFoundFalse(rName),
 		fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
@@ -1536,9 +1442,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_public_subnet_az1.id, aws_subnet.example_public_subnet_az2.id, aws_subnet.example_public_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1563,8 +1469,8 @@ resource "aws_msk_cluster" "test" {
 
 func testAccClusterConfig_brokerNodeGroupInfoPublicAccessSASLIAM(rName string, pa string) string {
 	return acctest.ConfigCompose(
-		testAccClusterBasePublicAccessConfig(rName),
-		testAccConfigurationAllowEveryoneNoACLFoundFalse(rName),
+		testAccClusterConfig_basePublicAccess(rName),
+		testAccClusterConfig_allowEveryoneNoACLFoundFalse(rName),
 		fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {
   cluster_name           = %[1]q
@@ -1572,9 +1478,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_public_subnet_az1.id, aws_subnet.example_public_subnet_az2.id, aws_subnet.example_public_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1632,9 +1538,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1674,9 +1580,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1703,9 +1609,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1733,9 +1639,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1772,9 +1678,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1817,9 +1723,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1852,9 +1758,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1878,9 +1784,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1906,9 +1812,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1935,9 +1841,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1958,9 +1864,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -1980,9 +1886,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = %[2]d
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -2002,9 +1908,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -2105,9 +2011,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -2153,9 +2059,9 @@ resource "aws_msk_cluster" "test" {
   }
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -2197,9 +2103,9 @@ resource "aws_msk_cluster" "test" {
   }
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -2224,9 +2130,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
@@ -2250,9 +2156,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.m5.large"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {
diff --git a/internal/service/kafka/scram_secret_association_test.go b/internal/service/kafka/scram_secret_association_test.go
index c8e30c212274..9569947dc58b 100644
--- a/internal/service/kafka/scram_secret_association_test.go
+++ b/internal/service/kafka/scram_secret_association_test.go
@@ -194,9 +194,9 @@ resource "aws_msk_cluster" "test" {
   number_of_broker_nodes = 3
 
   broker_node_group_info {
-    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    client_subnets  = aws_subnet.test[*].id
     instance_type   = "kafka.t3.small"
-    security_groups = [aws_security_group.example_sg.id]
+    security_groups = [aws_security_group.test.id]
 
     storage_info {
       ebs_storage_info {

From ed881ea8d77581e4544fd00afa853eb093116487 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 11 May 2023 07:58:53 -0400
Subject: [PATCH 4/7] Use 'extended_s3_configuration' in
 'aws_kinesis_firehose_delivery_stream' configuration.

---
 internal/service/kafka/cluster_test.go | 9 ++-------
 1 file changed, 2 insertions(+), 7 deletions(-)

diff --git a/internal/service/kafka/cluster_test.go b/internal/service/kafka/cluster_test.go
index b2327bce1be2..344981d563fe 100644
--- a/internal/service/kafka/cluster_test.go
+++ b/internal/service/kafka/cluster_test.go
@@ -1959,11 +1959,6 @@ resource "aws_s3_bucket" "bucket" {
   force_destroy = true
 }
 
-resource "aws_s3_bucket_acl" "test" {
-  bucket = aws_s3_bucket.bucket.id
-  acl    = "private"
-}
-
 resource "aws_iam_role" "firehose_role" {
   name = %[1]q
 
@@ -1986,9 +1981,9 @@ EOF
 
 resource "aws_kinesis_firehose_delivery_stream" "test" {
   name        = %[1]q
-  destination = "s3"
+  destination = "extended_s3"
 
-  s3_configuration {
+  extended_s3_configuration {
     role_arn   = aws_iam_role.firehose_role.arn
     bucket_arn = aws_s3_bucket.bucket.arn
   }

From e4070d83daf8c6ae8f9745e7b1381645c9a60222 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 11 May 2023 08:01:29 -0400
Subject: [PATCH 5/7] Add Terraform AWS Provider Version 5 Upgrade Guide entry.

---
 website/docs/guides/version-5-upgrade.html.md | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/website/docs/guides/version-5-upgrade.html.md b/website/docs/guides/version-5-upgrade.html.md
index 59819f7f9d9b..8b15f35a05f9 100644
--- a/website/docs/guides/version-5-upgrade.html.md
+++ b/website/docs/guides/version-5-upgrade.html.md
@@ -31,6 +31,7 @@ Upgrade topics:
 - [Resource: aws_docdb_cluster](#resource-aws_docdb_cluster)
 - [Resource: aws_ec2_client_vpn_endpoint](#resource-aws_ec2_client_vpn_endpoint)
 - [Resource: aws_ec2_client_vpn_network_association](#resource-aws_ec2_client_vpn_network_association)
+- [Resource: aws_msk_cluster](#resource-aws_msk_cluster)
 - [Resource: aws_neptune_cluster](#resource-aws_neptune_cluster)
 - [Resource: aws_rds_cluster](#resource-aws_rds_cluster)
 
@@ -129,6 +130,10 @@ The `status` attribute has been removed.
 
 The `status` attribute has been removed.
 
+## Resource: aws_msk_cluster
+
+The `broker_node_group_info.ebs_volume_size` attribute has been removed.
+
 ## Resource: aws_neptune_cluster
 
 Changes to the `snapshot_identifier` attribute will now correctly force re-creation of the resource. Previously, changing this attribute would result in a successful apply, but without the cluster being restored (only the resource state was changed). This change brings behavior of the cluster `snapshot_identifier` attribute into alignment with other RDS resources, such as `aws_db_instance`.

From 40c4627b611045b4aa609eb9e15c00ff381ad29d Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 11 May 2023 08:28:18 -0400
Subject: [PATCH 6/7] Fix Kendra documentation links.

---
 website/docs/d/kendra_index.html.markdown | 13 ++++++-------
 1 file changed, 6 insertions(+), 7 deletions(-)

diff --git a/website/docs/d/kendra_index.html.markdown b/website/docs/d/kendra_index.html.markdown
index 3777e1c25884..cd8ad09af06e 100644
--- a/website/docs/d/kendra_index.html.markdown
+++ b/website/docs/d/kendra_index.html.markdown
@@ -42,15 +42,14 @@ In addition to all of the arguments above, the following attributes are exported
 * `server_side_encryption_configuration` - A block that specifies the identifier of the AWS KMS customer managed key (CMK) that's used to encrypt data indexed by Amazon Kendra. Amazon Kendra doesn't support asymmetric CMKs. Documented below.
 * `status` - Current status of the index. When the value is `ACTIVE`, the index is ready for use. If the Status field value is `FAILED`, the `error_message` field contains a message that explains why.
 * `updated_at` - Unix datetime that the index was last updated.
-* `user_context_policy` - User context policy. Valid values are `ATTRIBUTE_FILTER` or `USER_TOKEN`. For more information, refer to [UserContextPolicy](https://docs.aws.amazon.com/kendra/latest/dg/API_CreateIndex.
-html#Kendra-CreateIndex-request-UserContextPolicy).
+* `user_context_policy` - User context policy. Valid values are `ATTRIBUTE_FILTER` or `USER_TOKEN`. For more information, refer to [UserContextPolicy](https://docs.aws.amazon.com/kendra/latest/APIReference/API_CreateIndex.html#kendra-CreateIndex-request-UserContextPolicy).
 * `user_group_resolution_configuration` - A block that enables fetching access levels of groups and users from an AWS Single Sign-On identity source. Documented below.
 * `user_token_configurations` - A block that specifies the user token configuration. Documented below.
 * `tags` - Metadata that helps organize the Indices you create.
 
 A `capacity_units` block supports the following attributes:
 
-* `query_capacity_units` - The amount of extra query capacity for an index and GetQuerySuggestions capacity. For more information, refer to [QueryCapacityUnits](https://docs.aws.amazon.com/kendra/latest/dg/API_CapacityUnitsConfiguration.html#Kendra-Type-CapacityUnitsConfiguration-QueryCapacityUnits).
+* `query_capacity_units` - The amount of extra query capacity for an index and GetQuerySuggestions capacity. For more information, refer to [QueryCapacityUnits](https://docs.aws.amazon.com/kendra/latest/APIReference/API_CapacityUnitsConfiguration.html#Kendra-Type-CapacityUnitsConfiguration-QueryCapacityUnits).
 * `storage_capacity_units` - The amount of extra storage capacity for an index. A single capacity unit provides 30 GB of storage space or 100,000 documents, whichever is reached first. Minimum value of 0.
 
 A `document_metadata_configuration_updates` block supports the following attributes:
@@ -62,11 +61,11 @@ A `document_metadata_configuration_updates` block supports the following attribu
 
 A `relevance` block supports the following attributes:
 
-* `duration` - Time period that the boost applies to. For more information, refer to [Duration](https://docs.aws.amazon.com/kendra/latest/dg/API_Relevance.html#Kendra-Type-Relevance-Duration).
-* `freshness` - How "fresh" a document is. For more information, refer to [Freshness](https://docs.aws.amazon.com/kendra/latest/dg/API_Relevance.html#Kendra-Type-Relevance-Freshness).
+* `duration` - Time period that the boost applies to. For more information, refer to [Duration](https://docs.aws.amazon.com/kendra/latest/APIReference/API_Relevance.html#Kendra-Type-Relevance-Duration).
+* `freshness` - How "fresh" a document is. For more information, refer to [Freshness](https://docs.aws.amazon.com/kendra/latest/APIReference/API_Relevance.html#Kendra-Type-Relevance-Freshness).
 * `importance` - Relative importance of the field in the search. Larger numbers provide more of a boost than smaller numbers. Minimum value of 1. Maximum value of 10.
-* `rank_order` - Determines how values should be interpreted. For more information, refer to [RankOrder](https://docs.aws.amazon.com/kendra/latest/dg/API_Relevance.html#Kendra-Type-Relevance-RankOrder).
-* `values_importance_map` - A list of values that should be given a different boost when they appear in the result list. For more information, refer to [ValueImportanceMap](https://docs.aws.amazon.com/kendra/latest/dg/API_Relevance.html#Kendra-Type-Relevance-ValueImportanceMap).
+* `rank_order` - Determines how values should be interpreted. For more information, refer to [RankOrder](https://docs.aws.amazon.com/kendra/latest/APIReference/API_Relevance.html#Kendra-Type-Relevance-RankOrder).
+* `values_importance_map` - A list of values that should be given a different boost when they appear in the result list. For more information, refer to [ValueImportanceMap](https://docs.aws.amazon.com/kendra/latest/APIReference/API_Relevance.html#Kendra-Type-Relevance-ValueImportanceMap).
 
 A `search` block supports the following attributes:
 

From 1d3ae40cffe576528a6faa88e8530e030c86b595 Mon Sep 17 00:00:00 2001
From: Kit Ewbank <Kit_Ewbank@hotmail.com>
Date: Thu, 11 May 2023 12:17:34 -0400
Subject: [PATCH 7/7] Fix golangci-lint 'unused'.

---
 internal/service/eks/addon_test.go     |  9 ---------
 internal/service/kafka/cluster_test.go | 21 ---------------------
 2 files changed, 30 deletions(-)

diff --git a/internal/service/eks/addon_test.go b/internal/service/eks/addon_test.go
index 5edc28eb573f..503e7c0a9b96 100644
--- a/internal/service/eks/addon_test.go
+++ b/internal/service/eks/addon_test.go
@@ -6,7 +6,6 @@ import (
 	"regexp"
 	"testing"
 
-	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/eks"
 	sdkacctest "github.com/hashicorp/terraform-plugin-testing/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-testing/helper/resource"
@@ -473,14 +472,6 @@ func testAccPreCheckAddon(ctx context.Context, t *testing.T) {
 	}
 }
 
-func testAccCheckAddonUpdateTags(ctx context.Context, addon *eks.Addon, oldTags, newTags map[string]string) resource.TestCheckFunc {
-	return func(s *terraform.State) error {
-		conn := acctest.Provider.Meta().(*conns.AWSClient).EKSConn()
-
-		return tfeks.UpdateTags(ctx, conn, aws.StringValue(addon.AddonArn), oldTags, newTags)
-	}
-}
-
 func testAccAddonConfig_base(rName string) string {
 	return acctest.ConfigCompose(acctest.ConfigAvailableAZsNoOptIn(), fmt.Sprintf(`
 data "aws_partition" "current" {}
diff --git a/internal/service/kafka/cluster_test.go b/internal/service/kafka/cluster_test.go
index 344981d563fe..e1da19d97ff3 100644
--- a/internal/service/kafka/cluster_test.go
+++ b/internal/service/kafka/cluster_test.go
@@ -1327,27 +1327,6 @@ resource "aws_msk_cluster" "test" {
 `, rName))
 }
 
-func testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeOnly(rName string, ebsVolumeSize int, instanceType string) string {
-	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
-resource "aws_msk_cluster" "test" {
-  cluster_name           = %[1]q
-  kafka_version          = "2.7.1"
-  number_of_broker_nodes = 3
-
-  broker_node_group_info {
-    client_subnets  = aws_subnet.test[*].id
-    instance_type   = %[3]q
-    security_groups = [aws_security_group.test.id]
-    storage_info {
-      ebs_storage_info {
-        volume_size = %[2]d
-      }
-    }
-  }
-}
-`, rName, ebsVolumeSize, instanceType))
-}
-
 func testAccClusterConfig_brokerNodeGroupInfoStorageInfoVolumeSizeSetAndProvThroughputNotEnabled(rName string, ebsVolumeSize int, instanceType string) string {
 	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
 resource "aws_msk_cluster" "test" {