feat: add topic_name_configuration to msk replicator

.changelog/40101.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_msk_replicator: Add `topic_name_configuration` argument
+```

internal/service/kafka/replicator.go

@@ -175,6 +175,22 @@ func resourceReplicator() *schema.Resource {
 											},
 										},
 									},
+									"topic_name_configuration": {
+										Type:     schema.TypeList,
+										Optional: true,
+										Computed: true,
+										MaxItems: 1,
+										Elem: &schema.Resource{
+											Schema: map[string]*schema.Schema{
+												names.AttrType: {
+													Type:             schema.TypeString,
+													Optional:         true,
+													ForceNew:         true,
+													ValidateDiagFunc: enum.Validate[types.ReplicationTopicNameConfigurationType](),
+												},
+											},
+										},
+									},
 									"topics_to_exclude": {
 										Type:     schema.TypeSet,
 										Optional: true,
@@ -579,6 +595,10 @@ func flattenTopicReplication(apiObject *types.TopicReplication) map[string]inter
 		tfMap["starting_position"] = []interface{}{flattenReplicationStartingPosition(v)}
 	}
 
+	if v := apiObject.TopicNameConfiguration; v != nil {
+		tfMap["topic_name_configuration"] = []interface{}{flattenReplicationTopicNameConfiguration(v)}
+	}
+
 	if v := apiObject.TopicsToReplicate; v != nil {
 		tfMap["topics_to_replicate"] = v
 	}
@@ -604,6 +624,20 @@ func flattenReplicationStartingPosition(apiObject *types.ReplicationStartingPosi
 	return tfMap
 }
 
+func flattenReplicationTopicNameConfiguration(apiObject *types.ReplicationTopicNameConfiguration) map[string]interface{} {
+	if apiObject == nil {
+		return nil
+	}
+
+	tfMap := map[string]interface{}{}
+
+	if v := apiObject.Type; v != "" {
+		tfMap[names.AttrType] = v
+	}
+
+	return tfMap
+}
+
 func flattenKafkaClusterDescriptions(apiObjects []types.KafkaClusterDescription) []interface{} { // nosemgrep:ci.kafka-in-func-name
 	if len(apiObjects) == 0 {
 		return nil
@@ -799,6 +833,10 @@ func expandTopicReplication(tfMap map[string]interface{}) *types.TopicReplicatio
 		apiObject.StartingPosition = expandReplicationStartingPosition(v[0].(map[string]interface{}))
 	}
 
+	if v, ok := tfMap["topic_name_configuration"].([]interface{}); ok && len(v) > 0 && v[0] != nil {
+		apiObject.TopicNameConfiguration = expandReplicationTopicNameConfiguration(v[0].(map[string]interface{}))
+	}
+
 	if v, ok := tfMap["topics_to_replicate"].(*schema.Set); ok && v.Len() > 0 {
 		apiObject.TopicsToReplicate = flex.ExpandStringValueSet(v)
 	}
@@ -820,6 +858,16 @@ func expandReplicationStartingPosition(tfMap map[string]interface{}) *types.Repl
 	return apiObject
 }
 
+func expandReplicationTopicNameConfiguration(tfMap map[string]interface{}) *types.ReplicationTopicNameConfiguration {
+	apiObject := &types.ReplicationTopicNameConfiguration{}
+
+	if v, ok := tfMap[names.AttrType].(string); ok {
+		apiObject.Type = types.ReplicationTopicNameConfigurationType(v)
+	}
+
+	return apiObject
+}
+
 func expandKafkaClusters(tfList []interface{}) []types.KafkaCluster { // nosemgrep:ci.kafka-in-func-name
 	if len(tfList) == 0 {
 		return nil

internal/service/kafka/replicator_test.go

@@ -56,6 +56,7 @@ func TestAccKafkaReplicator_basic(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.consumer_group_replication.0.consumer_groups_to_replicate.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.target_compression_type", "NONE"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.starting_position.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topic_name_configuration.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topics_to_replicate.#", "1"),
 				),
 			},
@@ -105,6 +106,7 @@ func TestAccKafkaReplicator_update(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.consumer_group_replication.0.consumer_groups_to_replicate.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.target_compression_type", "NONE"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.starting_position.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topic_name_configuration.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topics_to_replicate.#", "1"),
 				),
 			},
@@ -132,6 +134,8 @@ func TestAccKafkaReplicator_update(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.target_compression_type", "NONE"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.starting_position.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.starting_position.0.type", "EARLIEST"),
+					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topic_name_configuration.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topic_name_configuration.0.type", "IDENTICAL"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topics_to_replicate.#", "3"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.topics_to_exclude.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "replication_info_list.0.topic_replication.0.copy_topic_configurations", acctest.CtFalse),
@@ -629,6 +633,10 @@ resource "aws_msk_replicator" "test" {
       starting_position {
         type = "EARLIEST"
       }
+
+      topic_name_configuration {
+        type = "IDENTICAL"
+      }
     }
 
     consumer_group_replication {

website/docs/r/msk_replicator.html.markdown

@@ -49,6 +49,9 @@ resource "aws_msk_replicator" "test" {
 
 
     topic_replication {
+      topic_configuration_name {
+        type = "PREFIXED_WITH_SOURCE_CLUSTER_ALIAS"
+      }
       topics_to_replicate = [".*"]
       starting_position {
         type = "LATEST"
@@ -96,6 +99,7 @@ The following arguments are required:
 
 ### topic_replication Argument Reference
 
+* `topic_configuration_name` - (Optional) Configuration for specifying replicated topic names should be the same as their corresponding upstream topics or prefixed with source cluster alias.
 * `topics_to_replicate` - (Required) List of regular expression patterns indicating the topics to copy.
 * `topics_to_exclude` - (Optional) List of regular expression patterns indicating the topics that should not be replica.
 * `detect_and_copy_new_topics` - (Optional) Whether to periodically check for new topics and partitions.
@@ -110,6 +114,10 @@ The following arguments are required:
 * `detect_and_copy_new_consumer_groups` - (Optional) Whether to periodically check for new consumer groups.
 * `synchronise_consumer_group_offsets` - (Optional) Whether to periodically write the translated offsets to __consumer_offsets topic in target cluster.
 
+### topic_configuration_name
+
+* `type` - (optional) The type of topic configuration name. Supports `PREFIXED_WITH_SOURCE_CLUSTER_ALIAS` and `IDENTICAL`.
+
 ### starting_position
 
 * `type` - (Optional) The type of replication starting position. Supports `LATEST` and `EARLIEST`.