Skip to content

chore(kafka/instance): use new API to replace history API and improve acceptance test #8366

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Nov 26, 2025
Merged

chore(kafka/instance): use new API to replace history API and improve acceptance test #8366

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
354a360
chore(kafka): use new API to replace history API and improve test
wuzhuanhong Nov 19, 2025
7bc92f8
fix(kafka/instance): fix an error when updating storage_space
wuzhuanhong Nov 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ go 1.18

require (
github.com/GehirnInc/crypt v0.0.0-20200316065508-bb7000b8a962
github.com/chnsz/golangsdk v0.0.0-20251119014205-0fd4ae2d727b
github.com/chnsz/golangsdk v0.0.0-20251121031605-47043ab44ff7
github.com/google/go-cmp v0.6.0
github.com/hashicorp/go-cleanhttp v0.5.2
github.com/hashicorp/go-cty v1.4.1-0.20200414143053-d3edf31b6320
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,8 @@ github.com/apparentlymart/go-textseg/v13 v13.0.0 h1:Y+KvPE1NYz0xl601PVImeQfFyEy6
github.com/apparentlymart/go-textseg/v13 v13.0.0/go.mod h1:ZK2fH7c4NqDTLtiYLvIkEghdlcqw7yxLeM89kiTRPUo=
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
github.com/chnsz/golangsdk v0.0.0-20251119014205-0fd4ae2d727b h1:+5U4xfrnMRUQkteDBSClO+YHKa14xX+Fw9zHw/8+h7A=
github.com/chnsz/golangsdk v0.0.0-20251119014205-0fd4ae2d727b/go.mod h1:PIan4aDeDoNOI4FImVS8osVpShcVnAUKR2XM8Ulgsgw=
github.com/chnsz/golangsdk v0.0.0-20251121031605-47043ab44ff7 h1:CFHtM054TXBC5tBBKEAK+MAVCfsn/BtLFwQG0tJwxkM=
github.com/chnsz/golangsdk v0.0.0-20251121031605-47043ab44ff7/go.mod h1:PIan4aDeDoNOI4FImVS8osVpShcVnAUKR2XM8Ulgsgw=
github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
Expand Down
115 changes: 82 additions & 33 deletions huaweicloud/services/acceptance/kafka/resource_huaweicloud_dms_kafka_instance_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,24 +107,42 @@ func TestAccKafkaInstance_newFormat(t *testing.T) {
Check: resource.ComposeTestCheckFunc(
rc.CheckResourceExists(),
resource.TestCheckResourceAttr(resourceName, "name", rName),
resource.TestCheckResourceAttr(resourceName, "engine", "kafka"),
resource.TestCheckResourceAttr(resourceName, "security_protocol", "SASL_PLAINTEXT"),
resource.TestCheckResourceAttr(resourceName, "enabled_mechanisms.0", "SCRAM-SHA-512"),
resource.TestCheckResourceAttrPair(resourceName, "broker_num",
"data.huaweicloud_dms_kafka_flavors.test", "flavors.0.properties.0.min_broker"),
resource.TestCheckResourceAttrPair(resourceName, "flavor_id",
"data.huaweicloud_dms_kafka_flavors.test", "flavors.0.id"),
resource.TestCheckResourceAttr(resourceName, "engine_version", "2.7"),
resource.TestCheckResourceAttrPair(resourceName, "storage_spec_code",
"data.huaweicloud_dms_kafka_flavors.test", "flavors.0.ios.0.storage_spec_code"),
resource.TestCheckResourceAttr(resourceName, "broker_num", "3"),
resource.TestCheckResourceAttrPair(resourceName, "vpc_id",
"huaweicloud_vpc.test", "id"),
resource.TestCheckResourceAttrPair(resourceName, "security_group_id",
"huaweicloud_networking_secgroup.test", "id"),
resource.TestCheckResourceAttrPair(resourceName, "network_id",
"huaweicloud_vpc_subnet.test", "id"),
resource.TestMatchResourceAttr(resourceName, "availability_zones.#", regexp.MustCompile(`^[1-9]([0-9]*)?$`)),
resource.TestCheckResourceAttr(resourceName, "arch_type", "X86"),

resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.1.advertised_ip", "www.terraform-test.com"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.2.advertised_ip", "192.168.0.53"),
resource.TestCheckResourceAttrPair(resourceName, "flavor_id",
"data.huaweicloud_dms_kafka_flavors.test", "flavors.0.id"),
resource.TestCheckResourceAttr(resourceName, "broker_num", "3"),
resource.TestCheckResourceAttr(resourceName, "access_user", "user"),
resource.TestCheckResourceAttr(resourceName, "parameters.0.name", "log.retention.hours"),
resource.TestCheckResourceAttr(resourceName, "parameters.0.value", "48"),
resource.TestCheckResourceAttr(resourceName, "security_protocol", "SASL_PLAINTEXT"),
resource.TestCheckResourceAttr(resourceName, "enabled_mechanisms.0", "SCRAM-SHA-512"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.1.advertised_ip", "www.terraform-test.com"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.2.advertised_ip", "192.168.0.53"),
resource.TestCheckResourceAttrSet(resourceName, "storage_space"),
resource.TestCheckResourceAttrSet(resourceName, "maintain_begin"),
resource.TestCheckResourceAttrSet(resourceName, "maintain_end"),
// Check attributes.
resource.TestCheckResourceAttr(resourceName, "engine", "kafka"),
resource.TestMatchResourceAttr(resourceName, "partition_num", regexp.MustCompile(`^[1-9]([0-9]*)?$`)),
resource.TestCheckResourceAttrSet(resourceName, "connect_address"),
resource.TestCheckResourceAttrSet(resourceName, "port"),
resource.TestCheckResourceAttrSet(resourceName, "status"),
resource.TestCheckResourceAttrSet(resourceName, "storage_type"),
resource.TestCheckResourceAttrSet(resourceName, "created_at"),
resource.TestCheckResourceAttr(resourceName, "is_logical_volume", "true"),
resource.TestCheckResourceAttrSet(resourceName, "node_num"),
resource.TestCheckResourceAttrSet(resourceName, "pod_connect_address"),
resource.TestCheckResourceAttr(resourceName, "type", "cluster"),
),
},
{
Expand All @@ -139,13 +157,12 @@ func TestAccKafkaInstance_newFormat(t *testing.T) {
resource.TestCheckResourceAttrPair(resourceName, "storage_spec_code",
"data.huaweicloud_dms_kafka_flavors.test", "flavors.0.ios.0.storage_spec_code"),
resource.TestCheckResourceAttr(resourceName, "storage_space", "600"),

resource.TestCheckResourceAttr(resourceName, "parameters.0.name", "auto.create.groups.enable"),
resource.TestCheckResourceAttr(resourceName, "parameters.0.value", "false"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.0.advertised_ip", "192.168.0.61"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.1.advertised_ip", "test.terraform.com"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.2.advertised_ip", "192.168.0.62"),
resource.TestCheckResourceAttr(resourceName, "cross_vpc_accesses.3.advertised_ip", "192.168.0.63"),
resource.TestCheckResourceAttr(resourceName, "parameters.0.name", "auto.create.groups.enable"),
resource.TestCheckResourceAttr(resourceName, "parameters.0.value", "false"),
resource.TestCheckResourceAttr(resourceName, "maintain_begin", "06:00:00"),
resource.TestCheckResourceAttr(resourceName, "maintain_end", "10:00:00"),
),
Expand All @@ -166,7 +183,10 @@ func TestAccKafkaInstance_publicIp(t *testing.T) {
password := acceptance.RandomPassword()

resource.ParallelTest(t, resource.TestCase{
PreCheck: func() { acceptance.TestAccPreCheck(t) },
PreCheck: func() {
acceptance.TestAccPreCheck(t)
acceptance.TestAccPreCheckMigrateEpsID(t)
},
ProviderFactories: acceptance.TestAccProviderFactories,
CheckDestroy: rc.CheckResourceDestroy(),
Steps: []resource.TestStep{
Expand All @@ -177,13 +197,27 @@ func TestAccKafkaInstance_publicIp(t *testing.T) {
resource.TestCheckResourceAttr(resourceName, "name", rName),
resource.TestCheckResourceAttr(resourceName, "broker_num", "3"),
resource.TestCheckResourceAttr(resourceName, "public_ip_ids.#", "3"),
resource.TestCheckResourceAttr(resourceName, "dumping", "true"),
resource.TestCheckResourceAttr(resourceName, "new_tenant_ips.#", "2"),
resource.TestCheckResourceAttr(resourceName, "new_tenant_ips.0", "192.168.0.20"),
resource.TestCheckResourceAttr(resourceName, "new_tenant_ips.1", "192.168.0.18"),
resource.TestCheckResourceAttr(resourceName, "ssl_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.private_plain_enable", "true"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.public_plain_enable", "true"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.private_sasl_ssl_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.public_sasl_ssl_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.private_sasl_plaintext_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.public_sasl_plaintext_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "enable_auto_topic", "true"),
resource.TestCheckResourceAttr(resourceName, "vpc_client_plain", "true"),
resource.TestCheckResourceAttr(resourceName, "description", "Created by Terraform script"),
resource.TestCheckResourceAttr(resourceName, "enterprise_project_id", acceptance.HW_ENTERPRISE_PROJECT_ID_TEST),
resource.TestCheckResourceAttr(resourceName, "retention_policy", "time_base"),
// Check attributes.
resource.TestCheckResourceAttr(resourceName, "enable_public_ip", "true"),
resource.TestCheckResourceAttr(resourceName, "public_ip_address.#", "3"),
resource.TestCheckResourceAttrSet(resourceName, "connector_id"),
resource.TestCheckResourceAttrSet(resourceName, "connector_node_num"),
),
},
{
Expand All @@ -198,11 +232,17 @@ func TestAccKafkaInstance_publicIp(t *testing.T) {
resource.TestCheckResourceAttr(resourceName, "name", rName),
resource.TestCheckResourceAttr(resourceName, "broker_num", "4"),
resource.TestCheckResourceAttr(resourceName, "public_ip_ids.#", "4"),
resource.TestCheckResourceAttr(resourceName, "new_tenant_ips.#", "1"),
resource.TestCheckResourceAttr(resourceName, "new_tenant_ips.0", "192.168.0.79"),
resource.TestCheckResourceAttr(resourceName, "ssl_enable", "true"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.private_plain_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.private_sasl_ssl_enable", "true"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.public_plain_enable", "false"),
resource.TestCheckResourceAttr(resourceName, "port_protocol.0.public_sasl_ssl_enable", "true"),
resource.TestCheckResourceAttr(resourceName, "enable_auto_topic", "false"),
resource.TestCheckResourceAttr(resourceName, "description", ""),
resource.TestCheckResourceAttr(resourceName, "enterprise_project_id", acceptance.HW_ENTERPRISE_MIGRATE_PROJECT_ID_TEST),
resource.TestCheckResourceAttr(resourceName, "retention_policy", "produce_reject"),
),
},
{
Expand Down Expand Up @@ -479,18 +519,24 @@ resource "huaweicloud_dms_kafka_instance" "test" {
storage_spec_code = local.flavor.ios[0].storage_spec_code
availability_zones = slice(data.huaweicloud_availability_zones.test.names, 0, 3)

engine_version = "2.7"
storage_space = 300
broker_num = 3
arch_type = "X86"
public_ip_ids = huaweicloud_vpc_eip.test[*].id
new_tenant_ips = ["192.168.0.20", "192.168.0.18"]
engine_version = "2.7"
storage_space = 300
broker_num = 3
arch_type = "X86"
public_ip_ids = huaweicloud_vpc_eip.test[*].id
new_tenant_ips = ["192.168.0.20", "192.168.0.18"]
dumping = true
description = "Created by Terraform script"
enable_auto_topic = true
vpc_client_plain = true
enterprise_project_id = "%[4]s"
retention_policy = "time_base"

port_protocol {
private_plain_enable = true
public_plain_enable = true
}
}`, common.TestBaseNetwork(rName), testAccKafkaInstance_publicIpBase(3), rName)
}`, common.TestBaseNetwork(rName), testAccKafkaInstance_publicIpBase(3), rName, acceptance.HW_ENTERPRISE_PROJECT_ID_TEST)
}

func testAccKafkaInstance_publicIp_update(rName, password string, brokerNum int) string {
Expand Down Expand Up @@ -520,22 +566,25 @@ resource "huaweicloud_dms_kafka_instance" "test" {
storage_spec_code = local.flavor.ios[0].storage_spec_code
availability_zones = slice(data.huaweicloud_availability_zones.test.names, 0, 3)

engine_version = "2.7"
storage_space = 600
broker_num = %d
arch_type = "X86"
new_tenant_ips = ["192.168.0.79"]
public_ip_ids = huaweicloud_vpc_eip.test[*].id
access_user = "test"
password = "%[5]s"

enabled_mechanisms = ["SCRAM-SHA-512"]
engine_version = "2.7"
storage_space = 600
broker_num = %d
arch_type = "X86"
new_tenant_ips = ["192.168.0.79"]
dumping = true
public_ip_ids = huaweicloud_vpc_eip.test[*].id
access_user = "test"
password = "%[5]s"
enabled_mechanisms = ["SCRAM-SHA-512"]
enable_auto_topic = false
enterprise_project_id = "%[6]s"
retention_policy = "produce_reject"

port_protocol {
private_plain_enable = false
private_sasl_ssl_enable = true
public_plain_enable = false
public_sasl_ssl_enable = true
}
}`, common.TestBaseNetwork(rName), testAccKafkaInstance_publicIpBase(4), rName, brokerNum, password)
}`, common.TestBaseNetwork(rName), testAccKafkaInstance_publicIpBase(4), rName, brokerNum, password, acceptance.HW_ENTERPRISE_MIGRATE_PROJECT_ID_TEST)
}
24 changes: 19 additions & 5 deletions huaweicloud/services/kafka/resource_huaweicloud_dms_kafka_instance.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,11 +36,11 @@ const engineKafka = "kafka"

// @API Kafka GET /v2/available-zones
// @API Kafka POST /v2/{project_id}/instances/{instance_id}/crossvpc/modify
// @API Kafka POST /v2/{project_id}/instances/{instance_id}/extend
// @API Kafka POST /v2/{project_id}/kafka/instances/{instance_id}/extend
// @API Kafka DELETE /v2/{project_id}/instances/{instance_id}
// @API Kafka GET /v2/{project_id}/instances/{instance_id}
// @API Kafka PUT /v2/{project_id}/instances/{instance_id}
// @API Kafka POST /v2/{engine}/{project_id}/instances
// @API Kafka POST /v2/{project_id}/kafka/instances
// @API Kafka GET /v2/{project_id}/kafka/{instance_id}/tags
// @API Kafka POST /v2/{project_id}/kafka/{instance_id}/tags/action
// @API Kafka POST /v2/{project_id}/instances/{instance_id}/autotopic
Expand Down Expand Up @@ -919,7 +919,7 @@ func createKafkaInstanceWithFlavor(ctx context.Context, d *schema.ResourceData,
createOpts.Password = password
createOpts.KafkaManagerPassword = d.Get("manager_password").(string)

kafkaInstance, err := instances.CreateWithEngine(client, createOpts, engineKafka).Extract()
kafkaInstance, err := instances.CreateInstance(client, createOpts).Extract()
if err != nil {
return diag.Errorf("error creating Kafka instance: %s", err)
}
Expand Down Expand Up @@ -1063,7 +1063,7 @@ func createKafkaInstanceWithProductID(ctx context.Context, d *schema.ResourceDat
createOpts.Password = password
createOpts.KafkaManagerPassword = d.Get("manager_password").(string)

kafkaInstance, err := instances.CreateWithEngine(client, createOpts, engineKafka).Extract()
kafkaInstance, err := instances.CreateInstance(client, createOpts).Extract()
if err != nil {
return diag.Errorf("error creating DMS kafka instance: %s", err)
}
Expand Down Expand Up @@ -1729,7 +1729,21 @@ func resizeKafkaInstance(ctx context.Context, d *schema.ResourceData, meta inter
}
}

// `storage_space` equals `broker_num` multiplied by the storage space of each broker.
// The `storage_space` value is related to the `broker_num` value.
// After broker_num changes, we need to obtain the latest storage_space and compare it with the current storage_space.
if d.HasChanges("storage_space") {
resp, err := instances.Get(client, d.Id()).Extract()
if err != nil {
return fmt.Errorf("error getting Kafka instance: %s", err)
}

newStorageSpace := d.Get("storage_space").(int)
if resp.TotalStorageSpace >= newStorageSpace {
log.Printf("[WARN] The new storage space is less than or equal to the current storage space, no need to resize")
return nil
}

if err = resizeKafkaInstanceStorage(ctx, d, client); err != nil {
return err
}
Expand All @@ -1752,7 +1766,7 @@ func resizeKafkaInstanceStorage(ctx context.Context, d *schema.ResourceData, cli

func doKafkaInstanceResize(ctx context.Context, d *schema.ResourceData, client *golangsdk.ServiceClient, opts instances.ResizeInstanceOpts) error {
retryFunc := func() (interface{}, bool, error) {
_, err := instances.Resize(client, d.Id(), opts)
_, err := instances.ExtendInstance(client, d.Id(), opts)
retry, err := handleMultiOperationsError(err)
return nil, retry, err
}
Expand Down
37 changes: 37 additions & 0 deletions vendor/github.com/chnsz/golangsdk/openstack/dms/v2/kafka/instances/requests.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 8 additions & 0 deletions vendor/github.com/chnsz/golangsdk/openstack/dms/v2/kafka/instances/urls.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion vendor/modules.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ github.com/apparentlymart/go-cidr/cidr
# github.com/apparentlymart/go-textseg/v13 v13.0.0
## explicit; go 1.16
github.com/apparentlymart/go-textseg/v13/textseg
# github.com/chnsz/golangsdk v0.0.0-20251119014205-0fd4ae2d727b
# github.com/chnsz/golangsdk v0.0.0-20251121031605-47043ab44ff7
## explicit; go 1.14
github.com/chnsz/golangsdk
github.com/chnsz/golangsdk/auth
Expand Down
Loading