refactor tagging code to use keyvaluetags for aws msk resources & data source

aws/data_source_aws_msk_cluster.go

@@ -7,6 +7,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/kafka"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/validation"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/keyvaluetags"
 )
 
 func dataSourceAwsMskCluster() *schema.Resource {
@@ -103,7 +104,7 @@ func dataSourceAwsMskClusterRead(d *schema.ResourceData, meta interface{}) error
 	d.Set("kafka_version", aws.StringValue(cluster.CurrentBrokerSoftwareInfo.KafkaVersion))
 	d.Set("number_of_broker_nodes", aws.Int64Value(cluster.NumberOfBrokerNodes))
 
-	if err := d.Set("tags", tagsToMapMskCluster(cluster.Tags)); err != nil {
+	if err := d.Set("tags", keyvaluetags.KafkaKeyValueTags(cluster.Tags).IgnoreAws().Map()); err != nil {
 		return fmt.Errorf("error setting tags: %s", err)
 	}
 

aws/resource_aws_msk_cluster.go

@@ -10,6 +10,7 @@ import (
 	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/validation"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/keyvaluetags"
 )
 
 func resourceAwsMskCluster() *schema.Resource {
@@ -227,7 +228,7 @@ func resourceAwsMskClusterCreate(d *schema.ResourceData, meta interface{}) error
 		EnhancedMonitoring:   aws.String(d.Get("enhanced_monitoring").(string)),
 		KafkaVersion:         aws.String(d.Get("kafka_version").(string)),
 		NumberOfBrokerNodes:  aws.Int64(int64(d.Get("number_of_broker_nodes").(int))),
-		Tags:                 tagsFromMapMskCluster(d.Get("tags").(map[string]interface{})),
+		Tags:                 keyvaluetags.New(d.Get("tags").(map[string]interface{})).IgnoreAws().KafkaTags(),
 	}
 
 	out, err := conn.CreateCluster(input)
@@ -338,7 +339,7 @@ func resourceAwsMskClusterRead(d *schema.ResourceData, meta interface{}) error {
 	d.Set("kafka_version", aws.StringValue(cluster.CurrentBrokerSoftwareInfo.KafkaVersion))
 	d.Set("number_of_broker_nodes", aws.Int64Value(cluster.NumberOfBrokerNodes))
 
-	if err := d.Set("tags", tagsToMapMskCluster(cluster.Tags)); err != nil {
+	if err := d.Set("tags", keyvaluetags.KafkaKeyValueTags(cluster.Tags).IgnoreAws().Map()); err != nil {
 		return fmt.Errorf("error setting tags: %s", err)
 	}
 
@@ -404,8 +405,10 @@ func resourceAwsMskClusterUpdate(d *schema.ResourceData, meta interface{}) error
 	}
 
 	if d.HasChange("tags") {
-		if err := setTagsMskCluster(conn, d, d.Id()); err != nil {
-			return fmt.Errorf("failed updating tags for msk cluster %q: %s", d.Id(), err)
+		o, n := d.GetChange("tags")
+
+		if err := keyvaluetags.KafkaUpdateTags(conn, d.Id(), o, n); err != nil {
+			return fmt.Errorf("error updating MSK Cluster (%s) tags: %s", d.Id(), err)
 		}
 	}
 

aws/resource_aws_msk_cluster_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/hashicorp/terraform-plugin-sdk/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
 	"github.com/hashicorp/terraform-plugin-sdk/terraform"
+	"github.com/terraform-providers/terraform-provider-aws/aws/internal/keyvaluetags"
 )
 
 func init() {
@@ -508,7 +509,7 @@ func testAccLoadMskTags(cluster *kafka.ClusterInfo, td *kafka.ListTagsForResourc
 
 func testAccCheckMskClusterTags(td *kafka.ListTagsForResourceOutput, key string, value string) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
-		m := tagsToMapMskCluster(td.Tags)
+		m := keyvaluetags.KafkaKeyValueTags(td.Tags).IgnoreAws().Map()
 		v, ok := m[key]
 		if value != "" && !ok {
 			return fmt.Errorf("Missing tag: %s - (found tags %v)", key, m)