r/aws_mskconnect_connector: Allow connector_configuration to be updated in-place

.changelog/41685.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+resource/aws_mskconnect_connector: Allow `connector_configuration` to be updated in-place
+```

internal/service/kafkaconnect/connector.go

@@ -143,7 +143,6 @@ func resourceConnector() *schema.Resource {
 				Type:     schema.TypeMap,
 				Elem:     &schema.Schema{Type: schema.TypeString},
 				Required: true,
-				ForceNew: true,
 			},
 			names.AttrDescription: {
 				Type:         schema.TypeString,
@@ -517,20 +516,46 @@ func resourceConnectorUpdate(ctx context.Context, d *schema.ResourceData, meta i
 	conn := meta.(*conns.AWSClient).KafkaConnectClient(ctx)
 
 	if d.HasChangesExcept(names.AttrTags, names.AttrTagsAll) {
-		input := &kafkaconnect.UpdateConnectorInput{
-			Capacity:       expandCapacityUpdate(d.Get("capacity").([]interface{})[0].(map[string]interface{})),
-			ConnectorArn:   aws.String(d.Id()),
-			CurrentVersion: aws.String(d.Get(names.AttrVersion).(string)),
-		}
+		currentVersion := d.Get(names.AttrVersion).(string)
 
-		_, err := conn.UpdateConnector(ctx, input)
+		if d.HasChange("capacity") {
+			input := &kafkaconnect.UpdateConnectorInput{
+				Capacity:       expandCapacityUpdate(d.Get("capacity").([]interface{})[0].(map[string]interface{})),
+				ConnectorArn:   aws.String(d.Id()),
+				CurrentVersion: aws.String(currentVersion),
+			}
 
-		if err != nil {
-			return sdkdiag.AppendErrorf(diags, "updating MSK Connect Connector (%s): %s", d.Id(), err)
+			_, err := conn.UpdateConnector(ctx, input)
+
+			if err != nil {
+				return sdkdiag.AppendErrorf(diags, "updating MSK Connect Connector capacity (%s): %s", d.Id(), err)
+			}
+
+			output, err := waitConnectorUpdated(ctx, conn, d.Id(), d.Timeout(schema.TimeoutUpdate))
+
+			if err != nil {
+				return sdkdiag.AppendErrorf(diags, "waiting for MSK Connect Connector (%s) update: %s", d.Id(), err)
+			}
+
+			currentVersion = aws.ToString(output.CurrentVersion)
 		}
 
-		if _, err := waitConnectorUpdated(ctx, conn, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
-			return sdkdiag.AppendErrorf(diags, "waiting for MSK Connect Connector (%s) update: %s", d.Id(), err)
+		if d.HasChange("connector_configuration") {
+			input := &kafkaconnect.UpdateConnectorInput{
+				ConnectorConfiguration: flex.ExpandStringValueMap(d.Get("connector_configuration").(map[string]interface{})),
+				ConnectorArn:           aws.String(d.Id()),
+				CurrentVersion:         aws.String(currentVersion),
+			}
+
+			_, err := conn.UpdateConnector(ctx, input)
+
+			if err != nil {
+				return sdkdiag.AppendErrorf(diags, "updating MSK Connect Connector configuration (%s): %s", d.Id(), err)
+			}
+
+			if _, err := waitConnectorUpdated(ctx, conn, d.Id(), d.Timeout(schema.TimeoutUpdate)); err != nil {
+				return sdkdiag.AppendErrorf(diags, "waiting for MSK Connect Connector (%s) update: %s", d.Id(), err)
+			}
 		}
 	}
 

internal/service/kafkaconnect/connector_test.go

@@ -186,7 +186,7 @@ func TestAccKafkaConnectConnector_update(t *testing.T) {
 				ImportStateVerify: true,
 			},
 			{
-				Config: testAccConnectorConfig_allAttributesCapacityUpdated(rName),
+				Config: testAccConnectorConfig_allAttributesCapacityAndConnectorConfigUpdated(rName),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					testAccCheckConnectorExists(ctx, resourceName),
 					acctest.MatchResourceAttrRegionalARN(ctx, resourceName, names.AttrARN, "kafkaconnect", regexache.MustCompile(`connector/`+rName+`/`+kafkaConnectUUIDRegexPattern+`$`)),
@@ -198,7 +198,7 @@ func TestAccKafkaConnectConnector_update(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "connector_configuration.%", "3"),
 					resource.TestCheckResourceAttr(resourceName, "connector_configuration.connector.class", "com.github.jcustenborder.kafka.connect.simulator.SimulatorSinkConnector"),
 					resource.TestCheckResourceAttr(resourceName, "connector_configuration.tasks.max", "1"),
-					resource.TestCheckResourceAttr(resourceName, "connector_configuration.topics", "t1"),
+					resource.TestCheckResourceAttr(resourceName, "connector_configuration.topics", "t1, t2"),
 					resource.TestCheckResourceAttr(resourceName, names.AttrDescription, ""),
 					resource.TestCheckResourceAttr(resourceName, "kafka_cluster.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "kafka_cluster.0.apache_kafka_cluster.#", "1"),
@@ -579,7 +579,7 @@ resource "aws_mskconnect_connector" "test" {
 `, rName))
 }
 
-func testAccConnectorConfig_allAttributesCapacityUpdated(rName string) string {
+func testAccConnectorConfig_allAttributesCapacityAndConnectorConfigUpdated(rName string) string {
 	return acctest.ConfigCompose(
 		testAccCustomPluginConfig_basic(rName),
 		testAccWorkerConfigurationConfig_basic(rName),
@@ -603,7 +603,7 @@ resource "aws_mskconnect_connector" "test" {
   connector_configuration = {
     "connector.class" = "com.github.jcustenborder.kafka.connect.simulator.SimulatorSinkConnector"
     "tasks.max"       = "1"
-    "topics"          = "t1"
+    "topics"          = "t1, t2"
   }
 
   kafka_cluster {