Add support for brokerCapacityConfig for managedkafka (#15638) (#25074)

[upstream:abf05ffe9573613da8cdd17664204643770bf403]

Signed-off-by: Modular Magician <[email protected]>

Author: modular-magician

.changelog/15638.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+managedkafka: added `broker_capacity_config` field to `google_managed_kafka_cluster` resource
+```

google/services/managedkafka/resource_managed_kafka_cluster.go

@@ -183,6 +183,21 @@ func ResourceManagedKafkaCluster() *schema.Resource {
 				ForceNew:    true,
 				Description: `ID of the location of the Kafka resource. See https://cloud.google.com/managed-kafka/docs/locations for a list of supported locations.`,
 			},
+			"broker_capacity_config": {
+				Type:        schema.TypeList,
+				Optional:    true,
+				Description: `Capacity configuration at a per-broker level within the Kafka cluster. The config will be appled to each broker in the cluster.`,
+				MaxItems:    1,
+				Elem: &schema.Resource{
+					Schema: map[string]*schema.Schema{
+						"disk_size_gb": {
+							Type:        schema.TypeString,
+							Optional:    true,
+							Description: `The disk to provision for each broker in Gigabytes. Minimum: 100 GB.`,
+						},
+					},
+				},
+			},
 			"labels": {
 				Type:     schema.TypeMap,
 				Optional: true,
@@ -312,6 +327,12 @@ func resourceManagedKafkaClusterCreate(d *schema.ResourceData, meta interface{})
 	} else if v, ok := d.GetOkExists("capacity_config"); !tpgresource.IsEmptyValue(reflect.ValueOf(capacityConfigProp)) && (ok || !reflect.DeepEqual(v, capacityConfigProp)) {
 		obj["capacityConfig"] = capacityConfigProp
 	}
+	brokerCapacityConfigProp, err := expandManagedKafkaClusterBrokerCapacityConfig(d.Get("broker_capacity_config"), d, config)
+	if err != nil {
+		return err
+	} else if v, ok := d.GetOkExists("broker_capacity_config"); !tpgresource.IsEmptyValue(reflect.ValueOf(brokerCapacityConfigProp)) && (ok || !reflect.DeepEqual(v, brokerCapacityConfigProp)) {
+		obj["brokerCapacityConfig"] = brokerCapacityConfigProp
+	}
 	rebalanceConfigProp, err := expandManagedKafkaClusterRebalanceConfig(d.Get("rebalance_config"), d, config)
 	if err != nil {
 		return err
@@ -494,6 +515,12 @@ func resourceManagedKafkaClusterUpdate(d *schema.ResourceData, meta interface{})
 	} else if v, ok := d.GetOkExists("capacity_config"); !tpgresource.IsEmptyValue(reflect.ValueOf(v)) && (ok || !reflect.DeepEqual(v, capacityConfigProp)) {
 		obj["capacityConfig"] = capacityConfigProp
 	}
+	brokerCapacityConfigProp, err := expandManagedKafkaClusterBrokerCapacityConfig(d.Get("broker_capacity_config"), d, config)
+	if err != nil {
+		return err
+	} else if v, ok := d.GetOkExists("broker_capacity_config"); !tpgresource.IsEmptyValue(reflect.ValueOf(v)) && (ok || !reflect.DeepEqual(v, brokerCapacityConfigProp)) {
+		obj["brokerCapacityConfig"] = brokerCapacityConfigProp
+	}
 	rebalanceConfigProp, err := expandManagedKafkaClusterRebalanceConfig(d.Get("rebalance_config"), d, config)
 	if err != nil {
 		return err
@@ -530,6 +557,10 @@ func resourceManagedKafkaClusterUpdate(d *schema.ResourceData, meta interface{})
 		updateMask = append(updateMask, "capacityConfig")
 	}
 
+	if d.HasChange("broker_capacity_config") {
+		updateMask = append(updateMask, "brokerCapacityConfig")
+	}
+
 	if d.HasChange("rebalance_config") {
 		updateMask = append(updateMask, "rebalanceConfig")
 	}
@@ -976,6 +1007,32 @@ func expandManagedKafkaClusterCapacityConfigMemoryBytes(v interface{}, d tpgreso
 	return v, nil
 }
 
+func expandManagedKafkaClusterBrokerCapacityConfig(v interface{}, d tpgresource.TerraformResourceData, config *transport_tpg.Config) (interface{}, error) {
+	if v == nil {
+		return nil, nil
+	}
+	l := v.([]interface{})
+	if len(l) == 0 || l[0] == nil {
+		return nil, nil
+	}
+	raw := l[0]
+	original := raw.(map[string]interface{})
+	transformed := make(map[string]interface{})
+
+	transformedDiskSizeGb, err := expandManagedKafkaClusterBrokerCapacityConfigDiskSizeGb(original["disk_size_gb"], d, config)
+	if err != nil {
+		return nil, err
+	} else if val := reflect.ValueOf(transformedDiskSizeGb); val.IsValid() && !tpgresource.IsEmptyValue(val) {
+		transformed["diskSizeGb"] = transformedDiskSizeGb
+	}
+
+	return transformed, nil
+}
+
+func expandManagedKafkaClusterBrokerCapacityConfigDiskSizeGb(v interface{}, d tpgresource.TerraformResourceData, config *transport_tpg.Config) (interface{}, error) {
+	return v, nil
+}
+
 func expandManagedKafkaClusterRebalanceConfig(v interface{}, d tpgresource.TerraformResourceData, config *transport_tpg.Config) (interface{}, error) {
 	if v == nil {
 		return nil, nil

google/services/managedkafka/resource_managed_kafka_cluster_generated_meta.yaml

@@ -6,6 +6,7 @@ api_version: 'v1'
 api_resource_type_kind: 'Cluster'
 autogen_status: true
 fields:
+  - api_field: 'brokerCapacityConfig.diskSizeGb'
   - api_field: 'capacityConfig.memoryBytes'
   - api_field: 'capacityConfig.vcpuCount'
   - field: 'cluster_id'

google/services/managedkafka/resource_managed_kafka_cluster_generated_test.go

@@ -69,7 +69,7 @@ func TestAccManagedKafkaCluster_managedkafkaClusterBasicExample(t *testing.T) {
 				ResourceName:            "google_managed_kafka_cluster.example",
 				ImportState:             true,
 				ImportStateVerify:       true,
-				ImportStateVerifyIgnore: []string{"cluster_id", "labels", "location", "terraform_labels"},
+				ImportStateVerifyIgnore: []string{"broker_capacity_config", "cluster_id", "labels", "location", "terraform_labels"},
 			},
 		},
 	})
@@ -123,7 +123,7 @@ func TestAccManagedKafkaCluster_managedkafkaClusterMtlsExample(t *testing.T) {
 				ResourceName:            "google_managed_kafka_cluster.example",
 				ImportState:             true,
 				ImportStateVerify:       true,
-				ImportStateVerifyIgnore: []string{"cluster_id", "labels", "location", "terraform_labels"},
+				ImportStateVerifyIgnore: []string{"broker_capacity_config", "cluster_id", "labels", "location", "terraform_labels"},
 			},
 		},
 	})

google/services/managedkafka/resource_managed_kafka_cluster_test.go

@@ -51,7 +51,7 @@ func TestAccManagedKafkaCluster_update(t *testing.T) {
 				ResourceName:            "google_managed_kafka_cluster.example",
 				ImportState:             true,
 				ImportStateVerify:       true,
-				ImportStateVerifyIgnore: []string{"cluster_id", "labels", "location", "terraform_labels"},
+				ImportStateVerifyIgnore: []string{"broker_capacity_config", "cluster_id", "labels", "location", "terraform_labels"},
 			},
 			{
 				Config: testAccManagedKafkaCluster_updateTlsConfigToEmpty(context),
@@ -104,6 +104,9 @@ resource "google_managed_kafka_cluster" "example" {
     vcpu_count = 4
     memory_bytes = 4512135122
   }
+  broker_capacity_config {
+    disk_size_gb = 1500
+  }
   gcp_config {
     access_config {
       network_configs {

website/docs/r/managed_kafka_cluster.html.markdown

@@ -175,6 +175,11 @@ The following arguments are supported:
   **Note**: This field is non-authoritative, and will only manage the labels present in your configuration.
   Please refer to the field `effective_labels` for all of the labels present on the resource.
 
+* `broker_capacity_config` -
+  (Optional)
+  Capacity configuration at a per-broker level within the Kafka cluster. The config will be appled to each broker in the cluster.
+  Structure is [documented below](#nested_broker_capacity_config).
+
 * `rebalance_config` -
   (Optional)
   Defines rebalancing behavior of a Kafka cluster.
@@ -226,6 +231,12 @@ The following arguments are supported:
   (Required)
   The memory to provision for the cluster in bytes. The value must be between 1 GiB and 8 GiB per vCPU. Ex. 1024Mi, 4Gi.
 
+<a name="nested_broker_capacity_config"></a>The `broker_capacity_config` block supports:
+
+* `disk_size_gb` -
+  (Optional)
+  The disk to provision for each broker in Gigabytes. Minimum: 100 GB.
+
 <a name="nested_rebalance_config"></a>The `rebalance_config` block supports:
 
 * `mode` -