update: new data source aws_msk_bootstrap_brokers

.changelog/32484.txt

@@ -0,0 +1,3 @@
+```release-note:new-data-source
+aws_msk_bootstrap_brokers
+```

internal/service/kafka/bootstrap_brokers_data_source.go

@@ -0,0 +1,96 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package kafka
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-provider-aws/internal/conns"
+	"github.com/hashicorp/terraform-provider-aws/internal/errs/sdkdiag"
+	"github.com/hashicorp/terraform-provider-aws/internal/verify"
+)
+
+// @SDKDataSource("aws_msk_bootstrap_brokers", name="Bootstrap Brokers")
+func dataSourceBootstrapBrokers() *schema.Resource {
+	return &schema.Resource{
+		ReadWithoutTimeout: dataSourceBootstrapBrokersRead,
+
+		Schema: map[string]*schema.Schema{
+			"bootstrap_brokers": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_public_sasl_iam": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_public_sasl_scram": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_public_tls": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_sasl_iam": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_sasl_scram": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_tls": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_vpc_connectivity_sasl_iam": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_vpc_connectivity_sasl_scram": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"bootstrap_brokers_vpc_connectivity_tls": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
+			"cluster_arn": {
+				Type:         schema.TypeString,
+				Required:     true,
+				ValidateFunc: verify.ValidARN,
+			},
+		},
+	}
+}
+
+func dataSourceBootstrapBrokersRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	var diags diag.Diagnostics
+	conn := meta.(*conns.AWSClient).KafkaClient(ctx)
+
+	clusterARN := d.Get("cluster_arn").(string)
+	output, err := findBootstrapBrokersByARN(ctx, conn, clusterARN)
+
+	if err != nil {
+		return sdkdiag.AppendErrorf(diags, "reading MSK Cluster (%s) bootstrap brokers: %s", clusterARN, err)
+	}
+
+	d.SetId(clusterARN)
+	d.Set("bootstrap_brokers", SortEndpointsString(aws.ToString(output.BootstrapBrokerString)))
+	d.Set("bootstrap_brokers_public_sasl_iam", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringPublicSaslIam)))
+	d.Set("bootstrap_brokers_public_sasl_scram", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringPublicSaslScram)))
+	d.Set("bootstrap_brokers_public_tls", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringPublicTls)))
+	d.Set("bootstrap_brokers_sasl_iam", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringSaslIam)))
+	d.Set("bootstrap_brokers_sasl_scram", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringSaslScram)))
+	d.Set("bootstrap_brokers_tls", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringTls)))
+	d.Set("bootstrap_brokers_vpc_connectivity_sasl_iam", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringVpcConnectivitySaslIam)))
+	d.Set("bootstrap_brokers_vpc_connectivity_sasl_scram", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringVpcConnectivitySaslScram)))
+	d.Set("bootstrap_brokers_vpc_connectivity_tls", SortEndpointsString(aws.ToString(output.BootstrapBrokerStringVpcConnectivityTls)))
+
+	return diags
+}

internal/service/kafka/bootstrap_brokers_data_source_test.go

@@ -0,0 +1,74 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package kafka_test
+
+import (
+	"fmt"
+	"testing"
+
+	sdkacctest "github.com/hashicorp/terraform-plugin-testing/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-testing/helper/resource"
+	"github.com/hashicorp/terraform-provider-aws/internal/acctest"
+	"github.com/hashicorp/terraform-provider-aws/names"
+)
+
+func TestAccKafkaBootstrapBrokersDataSource_basic(t *testing.T) {
+	ctx := acctest.Context(t)
+	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
+	dataSourceName := "data.aws_msk_bootstrap_brokers.test"
+	resourceName := "aws_msk_cluster.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.PreCheck(ctx, t); testAccPreCheck(ctx, t) },
+		ErrorCheck:               acctest.ErrorCheck(t, names.KafkaEndpointID),
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccBootstrapBrokersDataSourceConfig_basic(rName),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers", resourceName, "bootstrap_brokers"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_public_sasl_iam", resourceName, "bootstrap_brokers_public_sasl_iam"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_public_sasl_scram", resourceName, "bootstrap_brokers_public_sasl_scram"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_public_tls", resourceName, "bootstrap_brokers_public_tls"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_sasl_iam", resourceName, "bootstrap_brokers_sasl_iam"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_sasl_scram", resourceName, "bootstrap_brokers_sasl_scram"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_tls", resourceName, "bootstrap_brokers_tls"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_vpc_connectivity_sasl_iam", resourceName, "bootstrap_brokers_vpc_connectivity_sasl_iam"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_vpc_connectivity_sasl_scram", resourceName, "bootstrap_brokers_vpc_connectivity_sasl_scram"),
+					resource.TestCheckResourceAttrPair(dataSourceName, "bootstrap_brokers_vpc_connectivity_tls", resourceName, "bootstrap_brokers_vpc_connectivity_tls"),
+				),
+			},
+		},
+	})
+}
+
+func testAccBootstrapBrokersDataSourceConfig_basic(rName string) string {
+	return acctest.ConfigCompose(testAccClusterConfig_base(rName), fmt.Sprintf(`
+resource "aws_msk_cluster" "test" {
+  cluster_name           = %[1]q
+  kafka_version          = "2.8.1"
+  number_of_broker_nodes = 3
+
+  broker_node_group_info {
+    client_subnets  = aws_subnet.test[*].id
+    instance_type   = "kafka.t3.small"
+    security_groups = [aws_security_group.test.id]
+
+    storage_info {
+      ebs_storage_info {
+        volume_size = 10
+      }
+    }
+  }
+
+  tags = {
+    Name = %[1]q
+  }
+}
+
+data "aws_msk_bootstrap_brokers" "test" {
+  cluster_arn = aws_msk_cluster.test.arn
+}
+`, rName))
+}

internal/service/kafka/broker_nodes_data_source.go

@@ -7,17 +7,18 @@ import (
 	"context"
 	"sort"
 
-	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/service/kafka"
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/kafka"
+	"github.com/aws/aws-sdk-go-v2/service/kafka/types"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
 	"github.com/hashicorp/terraform-provider-aws/internal/errs/sdkdiag"
 	"github.com/hashicorp/terraform-provider-aws/internal/verify"
 )
 
-// @SDKDataSource("aws_msk_broker_nodes")
-func DataSourceBrokerNodes() *schema.Resource {
+// @SDKDataSource("aws_msk_broker_nodes", name="Broker Nodes")
+func dataSourceBrokerNodes() *schema.Resource {
 	return &schema.Resource{
 		ReadWithoutTimeout: dataSourceBrokerNodesRead,
 
@@ -27,7 +28,6 @@ func DataSourceBrokerNodes() *schema.Resource {
 				Required:     true,
 				ValidateFunc: verify.ValidARN,
 			},
-
 			"node_info_list": {
 				Type:     schema.TypeList,
 				Computed: true,
@@ -67,53 +67,48 @@ func DataSourceBrokerNodes() *schema.Resource {
 
 func dataSourceBrokerNodesRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
 	var diags diag.Diagnostics
-	conn := meta.(*conns.AWSClient).KafkaConn(ctx)
+	conn := meta.(*conns.AWSClient).KafkaClient(ctx)
 
 	clusterARN := d.Get("cluster_arn").(string)
 	input := &kafka.ListNodesInput{
 		ClusterArn: aws.String(clusterARN),
 	}
-	var nodeInfos []*kafka.NodeInfo
+	var nodeInfos []types.NodeInfo
+
+	pages := kafka.NewListNodesPaginator(conn, input)
+	for pages.HasMorePages() {
+		page, err := pages.NextPage(ctx)
 
-	err := conn.ListNodesPagesWithContext(ctx, input, func(page *kafka.ListNodesOutput, lastPage bool) bool {
-		if page == nil {
-			return !lastPage
+		if err != nil {
+			return sdkdiag.AppendErrorf(diags, "listing MSK Cluster (%s) Broker Nodes: %s", clusterARN, err)
 		}
 
 		nodeInfos = append(nodeInfos, page.NodeInfoList...)
-
-		return !lastPage
-	})
-
-	if err != nil {
-		return sdkdiag.AppendErrorf(diags, "listing MSK Cluster (%s) Broker Nodes: %s", clusterARN, err)
 	}
 
 	// node list is returned unsorted sort on broker id
 	sort.Slice(nodeInfos, func(i, j int) bool {
-		iBrokerId := aws.Float64Value(nodeInfos[i].BrokerNodeInfo.BrokerId)
-		jBrokerId := aws.Float64Value(nodeInfos[j].BrokerNodeInfo.BrokerId)
+		iBrokerId := aws.ToFloat64(nodeInfos[i].BrokerNodeInfo.BrokerId)
+		jBrokerId := aws.ToFloat64(nodeInfos[j].BrokerNodeInfo.BrokerId)
 		return iBrokerId < jBrokerId
 	})
 
-	tfList := make([]interface{}, len(nodeInfos))
-
-	for i, apiObject := range nodeInfos {
-		brokerNodeInfo := apiObject.BrokerNodeInfo
-		tfMap := map[string]interface{}{
-			"attached_eni_id":       aws.StringValue(brokerNodeInfo.AttachedENIId),
-			"broker_id":             aws.Float64Value(brokerNodeInfo.BrokerId),
-			"client_subnet":         aws.StringValue(brokerNodeInfo.ClientSubnet),
-			"client_vpc_ip_address": aws.StringValue(brokerNodeInfo.ClientVpcIpAddress),
-			"endpoints":             aws.StringValueSlice(brokerNodeInfo.Endpoints),
-			"node_arn":              aws.StringValue(apiObject.NodeARN),
+	tfList := []interface{}{}
+	for _, apiObject := range nodeInfos {
+		if brokerNodeInfo := apiObject.BrokerNodeInfo; brokerNodeInfo != nil {
+			tfMap := map[string]interface{}{
+				"attached_eni_id":       aws.ToString(brokerNodeInfo.AttachedENIId),
+				"broker_id":             aws.ToFloat64(brokerNodeInfo.BrokerId),
+				"client_subnet":         aws.ToString(brokerNodeInfo.ClientSubnet),
+				"client_vpc_ip_address": aws.ToString(brokerNodeInfo.ClientVpcIpAddress),
+				"endpoints":             brokerNodeInfo.Endpoints,
+				"node_arn":              aws.ToString(apiObject.NodeARN),
+			}
+			tfList = append(tfList, tfMap)
 		}
-
-		tfList[i] = tfMap
 	}
 
 	d.SetId(clusterARN)
-
 	if err := d.Set("node_info_list", tfList); err != nil {
 		return sdkdiag.AppendErrorf(diags, "setting node_info_list: %s", err)
 	}

internal/service/kafka/broker_nodes_data_source_test.go

@@ -7,10 +7,10 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/aws/aws-sdk-go/service/kafka"
 	sdkacctest "github.com/hashicorp/terraform-plugin-testing/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-testing/helper/resource"
 	"github.com/hashicorp/terraform-provider-aws/internal/acctest"
+	"github.com/hashicorp/terraform-provider-aws/names"
 )
 
 func TestAccKafkaBrokerNodesDataSource_basic(t *testing.T) {
@@ -21,9 +21,8 @@ func TestAccKafkaBrokerNodesDataSource_basic(t *testing.T) {
 
 	resource.ParallelTest(t, resource.TestCase{
 		PreCheck:                 func() { acctest.PreCheck(ctx, t); testAccPreCheck(ctx, t) },
-		ErrorCheck:               acctest.ErrorCheck(t, kafka.EndpointsID),
+		ErrorCheck:               acctest.ErrorCheck(t, names.KafkaEndpointID),
 		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories,
-		CheckDestroy:             testAccCheckClusterDestroy(ctx),
 		Steps: []resource.TestStep{
 			{
 				Config: testAccBrokerNodesDataSourceConfig_basic(rName),