华为云用户手册

响应参数 状态码： 200 表3 响应Body参数 参数 参数类型 描述 total Integer 总条数 最小值：0 最大值：10000 producers Array of producers objects 生产者列表 数组长度：0 - 50 表4 producers 参数 参数类型 描述 producer_address String 生产者地址 最小长度：0 最大长度：64 broker_address String broker地址 最小长度：0 最大长度：64 join_time Long 加入时间 最小值：0 最大值：9223372036854775807

URI GET /v2/{project_id}/kafka/instances/{instance_id}/topics/{topic}/producers 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 最小长度：1 最大长度：64 instance_id 是 String 实例id 最小长度：1 最大长度：64 topic 是 String 主题 最小长度：1 最大长度：200 表2 Query参数 参数 是否必选 参数类型 描述 offset 否 Integer 偏移量，表示查询该偏移量后面的记录 最小值：0 最大值：10000 limit 否 Integer 查询返回记录的数量限制 最小值：1 最大值：50

响应示例 状态码： 200 查询topic的当前生产者列表成功 { "total" : 3, "producers" : [ { "producer_address" : "192.0.0.149:40443", "broker_address" : "192.0.0.146:9092", "join_time" : 1687204743328 }, { "producer_address" : "192.0.0.149:13807", "broker_address" : "192.0.0.80:9092", "join_time" : 1687204745939 }, { "producer_address" : "192.0.0.149:31876", "broker_address" : "192.0.0.71:9092", "join_time" : 1687204744934 } ] }

响应示例 状态码： 200 查询成功。 { "messages" : [ { "topic" : "topic-test", "partition" : 0, "value" : "hello world", "size" : 21, "timestamp" : 1607598463502, "huge_message" : false, "message_offset" : 4, "message_id" : "", "app_id" : "", "tag" : "" } ], "total" : 1, "size" : 1 }

请求示例 查询消息偏移量。 GET https://{endpoint}/v2/{project_id}/instances/{instance_id}/messages?asc=false&end_time=1608609032042&limit=10&offset=0&start_time=1608608432042&topic=topic-test 查询消息内容。 GET https://{endpoint}/v2/{project_id}/instances/{instance_id}/messages?download=false&message_offset=0&partition=0&topic=topic-test

响应参数 状态码： 200 表3 响应Body参数 参数 参数类型 描述 messages Array of MessagesEntity objects 消息列表。 total Long 消息总条数。 size Long 每页消息条数。 表4 MessagesEntity 参数 参数类型 描述 topic String topic名称。 partition Integer 消息所在的分区。 key String 消息key。 value String 消息内容。 size Integer 消息大小。 timestamp Long 生产消息的时间。 格式为Unix时间戳。单位为毫秒。 huge_message Boolean 大数据标识。 message_offset Long 消息偏移量。 message_id String 消息ID。 app_id String 应用ID。 tag String 消息标签。 状态码： 400 表5 响应Body参数 参数 参数类型 描述 error_code String 错误码。 error_msg String 错误描述。 状态码： 403 表6 响应Body参数 参数 参数类型 描述 error_code String 错误码。 error_msg String 错误描述。

URI GET /v2/{project_id}/instances/{instance_id}/messages 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 instance_id 是 String 实例ID。 表2 Query参数 参数 是否必选 参数类型 描述 topic 是 String Topic名称。 Topic名称必现以字母开头且只支持大小写字母、中横线、下划线以及数字。 asc 否 Boolean 是否按照时间排序。 start_time 否 String 开始时间。 Unix毫秒时间戳。 查询消息偏移量时，为必选参数。 end_time 否 String 结束时间。 Unix毫秒时间戳。 查询消息偏移量时，为必选参数。 limit 否 String 每一页显示的message数量。 offset 否 String 页数。 download 否 Boolean 是否下载。 message_offset 否 String 消息偏移量。 查询消息内容时，为必选参数。 若start_time、end_time参数不为空，该参数无效。 partition 否 String 分区。 查询消息内容时，为必选参数。 若start_time、end_time参数不为空，该参数无效。 keyword 否 String 关键词。 取值范围为0~50。 最小长度：0 最大长度：50

功能介绍 在创建kafka实例时，需要配置订购的产品ID（即product_id），可通过该接口查询产品规格。 例如，要订购按需计费、基准带宽为100MB的kafka实例，可从接口响应消息中，查找Hourly的消息体，然后找到bandwidth为100MB的记录对应的product_id，该product_id的值即是创建上述kafka实例时需要配置的产品ID。 同时，unavailable_zones字段表示资源不足的可用区列表，如果为空，则表示所有可用区都有资源，如果不为空，则表示字段值的可用区没有资源。所以必须确保您购买的资源所在的可用区有资源，不在该字段列表内。

响应消息 响应参数 Hourly或者Monthly的参数说明见表3。 表2 参数说明 参数 类型 说明 Hourly Array 表示按需付费的产品列表。 Monthly Array 表示包年包月的产品列表。当前暂不支持通过API创建包年包月的kafka实例。 表3 参数说明 参数 类型 说明 name String 消息引擎的名称，该字段显示为kafka。 version String 消息引擎的版本，当前支持1.1.0、2.3.0和2.7。 values Array 产品规格列表。具体参数，请参见表4。 表4 values参数说明 参数 类型 说明 detail Array 规格详情。具体参数，请参见表5。 name String 实例类型。 unavailable_zones Array 资源售罄的可用区。 available_zones Array 有可用资源的可用区。 表5 Kafka实例的detail参数说明 参数 类型 说明 tps String 单位时间内的消息量最大值。 storage String 消息存储空间。 partition_num String Kafka实例的最大Topic数。 product_id String 产品ID。 spec_code String 规格ID。 io Array IO信息。具体参数，请参见表6。 bandwidth String Kafka实例的基准带宽。 available_zones Array 实例资源未售罄的可用区。 ecs_flavor_id String 该产品规格对应的虚拟机规格。 arch_type String 实例规格架构类型。当前仅支持X86。 表6 io参数说明 参数 类型 说明 io_type String IO类型。 storage_spec_code String IO规格。 available_zones Array IO未售罄的可用区。 unavailable_zones Array of strings IO已售罄的不可用区列表。 volume_type String 磁盘类型。 响应示例 { "Hourly": [{ "name": "kafka", "version": "XXX", "values": [{ "detail": [{ "tps": "50000", "storage": "600", "partition_num": "300", "product_id": "00300-30308-0--0", "spec_code": "dms.instance.kafka.cluster.c3.mini", "io": [{ "io_type": "high", "storage_spec_code": "dms.physical.storage.high", "volume_type": "SAS" }, { "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "100MB", "unavailable_zones": [], "ecs_flavor_id": "c3.large.2" }, { "tps": "100000", "storage": "1200", "partition_num": "900", "product_id": "00300-30310-0--0", "spec_code": "dms.instance.kafka.cluster.c3.small.2", "io": [{ "io_type": "high", "storage_spec_code": "dms.physical.storage.high", "volume_type": "SAS" }, { "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "300MB", "unavailable_zones": [], "ecs_flavor_id": "c3.xlarge.2" }, { "tps": "200000", "storage": "2400", "partition_num": "1800", "product_id": "00300-30312-0--0", "spec_code": "dms.instance.kafka.cluster.c3.middle.2", "io": [{ "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "600MB", "unavailable_zones": [], "ecs_flavor_id": "c3.2xlarge.2" }, { "tps": "300000", "storage": "4800", "partition_num": "1800", "product_id": "00300-30314-0--0", "spec_code": "dms.instance.kafka.cluster.c3.high.2", "io": [{ "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "1200MB", "unavailable_zones": [], "ecs_flavor_id": "c3ne.2xlarge.2" }], "name": "cluster", "unavailable_zones": [] }] }], "Monthly": [{ "name": "kafka", "version": "XXX", "values": [{ "detail": [{ "tps": "50000", "storage": "600", "partition_num": "300", "product_id": "00300-30309-0--0", "spec_code": "dms.instance.kafka.cluster.c3.mini", "io": [{ "io_type": "high", "storage_spec_code": "dms.physical.storage.high", "volume_type": "SAS" }, { "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "100MB", "unavailable_zones": [], "ecs_flavor_id": "c3.large.2" }, { "tps": "100000", "storage": "1200", "partition_num": "900", "product_id": "00300-30311-0--0", "spec_code": "dms.instance.kafka.cluster.c3.small.2", "io": [{ "io_type": "high", "storage_spec_code": "dms.physical.storage.high", "volume_type": "SAS" }, { "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "300MB", "unavailable_zones": [], "ecs_flavor_id": "c3.xlarge.2" }, { "tps": "200000", "storage": "2400", "partition_num": "1800", "product_id": "00300-30313-0--0", "spec_code": "dms.instance.kafka.cluster.c3.middle.2", "io": [{ "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "600MB", "unavailable_zones": [], "ecs_flavor_id": "c3.2xlarge.2" }, { "tps": "300000", "storage": "4800", "partition_num": "1800", "product_id": "00300-30315-0--0", "spec_code": "dms.instance.kafka.cluster.c3.high.2", "io": [{ "io_type": "ultra", "storage_spec_code": "dms.physical.storage.ultra", "volume_type": "SSD" }], "bandwidth": "1200MB", "unavailable_zones": [], "ecs_flavor_id": "c3ne.2xlarge.2" }], "name": "cluster", "unavailable_zones": [] }] }] }

URI GET /v1.0/{project_id}/instances/{instance_id}/manage/topics/{topic}/messages?partition={partition}&start_time={start_time}&end_time={end_time}&limit={limit}&offset={offset} 参数说明请参考表1 表1 参数说明 参数 类型 必选 说明 project_id String 是 项目ID。 instance_id String 是 实例ID。 topic String 是 Topic名称。 partition Integer 否 分区编号，默认值为-1，若传入值为-1，则查询所有分区。 start_time Long 否 查询起始时间，为unix时间戳格式，默认值为0。 end_time Long 否 查询结束时间，为unix时间戳格式，默认值为系统当前时间。 limit Integer 否 单页返回消息数，默认值为10。 offset Integer 否 查询的页码，默认值为1。

响应参数 状态码： 200 表3 响应Body参数 参数 参数类型 描述 messages Array of messages objects 消息列表。 messages_count Integer 消息总数。 offsets_count Integer 总页数。 offset Integer 当前页数。 表4 messages 参数 参数类型 描述 topic String topic名称。 partition Integer 分区编号。 message_offset Long 消息编号。 size Integer 消息大小，单位字节。 timestamp Long 生产消息的时间。 格式为Unix时间戳。单位为毫秒。

响应示例 状态码： 200 查询分区指定时间段的消息成功。 { "messages" : [ { "topic" : "mytest", "partition" : 0, "message_offset" : 7, "size" : 6, "timestamp" : 1568125036045 } ], "messages_count" : 1, "offsets_count" : 1, "offset" : 1 }

URI GET /v2/{project_id}/instances/{instance_id}/management/topics/{topic}/messages 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 instance_id 是 String 实例ID。 topic 是 String Topic名称。 Topic名称必现以字母开头且只支持大小写字母、中横线、下划线以及数字。 表2 Query参数 参数 是否必选 参数类型 描述 start_time 否 String 查询起始时间，为unix时间戳格式，默认值为0。 end_time 否 String 查询结束时间，为unix时间戳格式，默认值为系统当前时间。 limit 否 Integer 单页返回消息数，默认值为10。 offset 否 Integer 偏移量，表示从此偏移量开始查询， offset大于等于0。 partition 否 String 分区编号，默认值为-1，若传入值为-1，则查询所有分区。

响应参数 状态码： 200 表2 响应Body参数 参数 参数类型 描述 tags Array of TagEntity objects 标签列表 表3 TagEntity 参数 参数类型 描述 key String 标签键。 不能为空。 对于同一个实例，Key值唯一。 长度为1~128个字符（中文也可以输入128个字符）。 由任意语种字母、数字、空格和字符组成，字符仅支持_ . : = + - @ 首尾字符不能为空格。 value String 标签值。 长度为0~255个字符（中文也可以输入255个字符）。 由任意语种字母、数字、空格和字符组成，字符仅支持_ . : = + - @ 首尾字符不能为空格。

响应示例 状态码： 200 查询实例的后台任务列表 { "task_count" : "4", "tasks" : [ { "id" : "8abfa7b372160bfd0172165864064079", "name" : "modifyAutoTopic", "user_name" : "paas_dms", "user_id" : "3df5acbc24a54fadb62a043c9000a307", "params" : "{\"old_auto_status\":true,\"new_auto_status\":false}", "status" : "EXECUTING", "created_at" : "2020-05-15T03:19:51.046Z", "updated_at" : "2020-05-15T03:19:51.065Z" }, { "id" : "8abfa7b372160bfd017216560af83e6e", "name" : "changeRetentionPolicy", "user_name" : "paas_dms", "user_id" : "3df5acbc24a54fadb62a043c9000a307", "params" : "{\"new_retention_policy\":\"produce_reject\",\"origin_retention_policy\":\"time_base\"}", "status" : "SUC CES S", "created_at" : "2020-05-15T03:17:17.176Z", "updated_at" : "2020-05-15T03:17:22.162Z" } ] }

URI GET /v2/{project_id}/instances/{instance_id}/tasks 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 instance_id 是 String 实例ID。 表2 Query参数 参数 是否必选 参数类型 描述 start 否 Integer 开启查询的任务编号。 limit 否 Integer 查询的任务个数。 begin_time 否 String 查询任务的最小时间，格式为YYYYMMDDHHmmss。 end_time 否 String 查询任务的最大时间，格式为YYYYMMDDHHmmss。

响应参数 状态码： 200 表3 响应Body参数 参数 参数类型 描述 task_count String 任务数量。 tasks Array of tasks objects 任务列表。 表4 tasks 参数 参数类型 描述 id String 任务ID。 name String 任务名称。 user_name String 用户名。 user_id String 用户ID。 params String 任务参数。 status String 任务状态。 created_at String 启动时间。 updated_at String 结束时间。

URI GET /v2/{project_id}/instances/{instance_id}/management/topics/{topic}/partitions/{partition}/beginning-message 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 instance_id 是 String 实例ID。 topic 是 String Topic名称。 Topic名称必现以字母开头且只支持大小写字母、中横线、下划线以及数字。 partition 是 Integer 分区编号。

响应消息 响应参数 当参数action为delete，allFailure值为kafka时，响应返回为空表示删除成功。参数说明见表3。 表3 参数说明 参数 类型 说明 results Array 修改实例的结果。 表4 results参数说明 参数 类型 说明 instance String 实例ID。 result String 操作结果：success、failed。 响应示例 { "results": [ { "result": "success", "instance": "afc90a2a-a02c-4cba-94d5-58dfa9ad1e0d" }, { "result": "success", "instance": "67fc5f8d-3986-4f02-bb75-4075a23112de" } ] }

请求消息 请求参数 参数说明见表2。 表2 参数说明 参数 类型 必选 说明 action String 是 对实例的操作：restart、delete instances Array 否 实例的ID列表。 allFailure String 否 参数值为kafka，表示删除租户所有创建失败的Kafka实例。 请求示例 批量重启实例 POST https://{dms_endpoint}/v1.0/{project_id}/instances/action { "action" : "restart", "instances" : ["54602a9d-5e22-4239-9123-77e350df4a34", "7166cdea-dbad-4d79-9610-7163e6f8b640"] } 批量删除实例 POST https://{dms_endpoint}/v1.0/{project_id}/instances/action { "action" : "delete", "instances" : ["54602a9d-5e22-4239-9123-77e350df4a34", "7166cdea-dbad-4d79-9610-7163e6f8b640"] } 删除所有创建失败的实例 POST https://{dms_endpoint}/v1.0/{project_id}/instances/action { "action" : "delete", "allFailure" : "kafka" }

响应示例 状态码： 200 查询成功。 { "task_count" : "1", "tasks" : [ { "id" : "8abfa7b272adc5b40172b73130065ae7", "name" : "bindInstancePublicIp", "user_name" : "paas_dms", "user_id" : "3df5acbc24a54fadb62a043c9000a307", "params" : "{\"public_ip_id\":\"1aea7aed-e7d8-40ea-b3de-6f3ee9d5db9f\",\"public_ip_address\":\"100.93.2.18\",\"enable_public_ip\":true}", "status" : "SUCCESS", "created_at" : "2020-06-15T08:55:53.606Z", "updated_at" : "2020-06-15T08:55:56.600Z" } ] }

响应参数 状态码： 200 表2 响应Body参数 参数 参数类型 描述 task_count String 任务数量。 tasks Array of tasks objects 任务列表。 表3 tasks 参数 参数类型 描述 id String 任务ID。 name String 任务名称。 user_name String 用户名。 user_id String 用户ID。 params String 任务参数。 status String 任务状态。 created_at String 启动时间。 updated_at String 结束时间。

URI POST /v1.0/{project_id}/instances/{instance_id}/manage/topics/{topic}/partitions-reassignment 参数说明如下表所示。 表1 参数说明 参数 类型 必选 说明 project_id String 是 项目ID。 instance_id String 是 实例ID。 topic String 是 Topic名称。

API概览 表1 实例管理类接口 API 说明 生命周期管理 包括创建实例、修改实例信息、查询实例、删除实例、重启实例。 实例管理 包括重置密码、重置Manager密码、重启Manager、开启或关闭实例自动创建Topic功能、修改实例跨VPC访问的内网IP、查询Kafka集群元数据信息、查询消费组信息、重置消费组消费进度到指定位置、查询Kafka实例的协调器信息、修改分区的副本、查询Topic的磁盘存储情况、查询所有消费组、查询指定消费组信息、删除消费组、Kafka实例开始分区重平衡任务。 转储管理 包括创建实例的转储节点、创建转储任务、查询转储任务列表、修改转储任务的配额、查询单个转储任务、删除单个转储任务。 规格变更管理 包括查询实例的扩容规格列表、实例扩容。 主题管理 包括查询Topic、创建Topic、修改Topic、批量删除Topic、查询Topic的分区列表、查询Topic的当前生产者列表、查询Topic详细信息。 用户管理 包括查询用户列表、创建用户、删除用户、重置用户密码、查询用户权限、设置用户权限。 消息查询 包括查询消息、查询分区指定偏移量的消息、查询分区指定时间段的消息、查询分区最早/最新的消息位置。 后台任务管理 包括查询实例的后台任务列表、查询后台任务管理中的指定记录、删除后台任务管理中的指定记录。 标签管理 包括批量添加或删除实例标签、查询实例标签、查询项目标签。 其他接口 包括查询产品规格列表、查询可用区信息、查询维护时间窗、查询实例在CES的监控层级关系。

响应示例 状态码： 200 查询成功。 { "broker_list" : [ { "broker_name" : "broker-0", "data_disk_size" : "66G", "data_disk_use" : "53M", "data_disk_free" : "63G", "data_disk_use_percentage" : "1", "status" : "Success get info", "topic_list" : [ { "size" : "12K", "topic_name" : "topic-test", "topic_partition" : "2", "percentage" : 1.7339533025568183E-5 }, { "size" : "12K", "topic_name" : "__consumer_offsets", "topic_partition" : "4", "percentage" : 1.7339533025568183E-5 }, { "size" : "12K", "topic_name" : "__consumer_offsets", "topic_partition" : "3", "percentage" : 1.7339533025568183E-5 }, { "size" : "8.0K", "topic_name" : "__trace", "topic_partition" : "6", "percentage" : 1.1559688683712121E-5 }, { "size" : "8.0K", "topic_name" : "__trace", "topic_partition" : "4", "percentage" : 1.1559688683712121E-5 }, { "size" : "8.0K", "topic_name" : "__trace", "topic_partition" : "2", "percentage" : 1.1559688683712121E-5 }, { "size" : "8.0K", "topic_name" : "__trace", "topic_partition" : "0", "percentage" : 1.1559688683712121E-5 }, { "size" : "8.0K", "topic_name" : "topic-test", "topic_partition" : "0", "percentage" : 1.1559688683712121E-5 }, { "size" : "8.0K", "topic_name" : "topic-1568537362", "topic_partition" : "2", "percentage" : 1.1559688683712121E-5 }, { "size" : "8.0K", "topic_name" : "__consumer_offsets", "topic_partition" : "7", "percentage" : 1.1559688683712121E-5 } ] } ] }

响应参数 状态码： 200 表3 响应Body参数 参数 参数类型 描述 broker_list Array of DiskusageEntity objects Broker列表。 表4 DiskusageEntity 参数 参数类型 描述 broker_name String Broker名称。 data_disk_size String 磁盘容量。 data_disk_use String 已使用的磁盘容量。 data_disk_free String 剩余可用的磁盘容量。 data_disk_use_percentage String 消息标签。 status String 消息标签。 topic_list Array of DiskusageTopicEntity objects topic磁盘容量使用列表。 表5 DiskusageTopicEntity 参数 参数类型 描述 size String 磁盘使用量。 topic_name String topic名称。 topic_partition String 分区。 percentage Double 磁盘使用量的占比。

URI GET /v2/{project_id}/instances/{instance_id}/topics/diskusage 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 instance_id 是 String 实例ID。 表2 Query参数 参数 是否必选 参数类型 描述 minSize 否 String 占用磁盘大小，默认值1G (1K，1M，1G)。 top 否 String 占用磁盘大小，查询top N。 percentage 否 String 占用磁盘大小，查询大于占比的分区。

URI DELETE /v2/{project_id}/connectors/{connector_id}/sink-tasks/{task_id} 表1 路径参数 参数 是否必选 参数类型 描述 project_id 是 String 项目ID，获取方式请参见获取项目ID。 connector_id 是 String 实例转储ID。 请参考查询实例返回的数据。 task_id 是 String 转储任务ID。

概述 欢迎使用分布式消息服务Kafka版。分布式消息服务Kafka版是一款基于开源社区版Kafka提供的消息队列服务，向用户提供计算、存储和带宽资源独占式的Kafka专享实例。使用分布式消息服务Kafka版，资源按需申请，按需配置Topic的分区与副本数量，即买即用，您将有更多精力专注于业务快速开发，不用考虑部署和运维。 本文档提供了分布式消息服务Kafka版API的描述、语法、参数说明及样例等内容。 分布式消息服务Kafka版持续增加新的功能，将不可避免对现有接口有所调整，比如增加响应参数。 为了减少接口变更带来的影响，除了分布式消息服务Kafka版自身尽量做到接口向下兼容的同时，用户在使用过程中，应当接受出现返回内容（JSON格式）含有未使用的参数和值的现象，即能够正常忽略未使用的参数和值。 父主题： 使用前必读

请求示例 扩容存储空间（按需实例）。 POST https://{endpoint}/v2/{engine}/{project_id}/instances/{instance_id}/extend { "oper_type" : "storage", "new_storage_space" : 600 } 扩容代理数量（按需实例）。 POST https://{endpoint}/v2/{engine}/{project_id}/instances/{instance_id}/extend { "oper_type" : "horizontal", "new_storage_space" : 1600, "new_broker_num" : 4, "tenant_ips" : [ "127.0.0.1", "127.0.0.2", "127.0.0.3" ] } 扩容代理规格（按需实例）。 POST https://{endpoint}/v2/{engine}/{project_id}/instances/{instance_id}/extend { "oper_type" : "vertical", "new_product_id" : "c6.4u8g.cluster" }