更新时间:2025-08-15 GMT+08:00

Kafka实例创建Topic

功能介绍

该接口用于向Kafka实例创建Topic。

调用方法

请参见如何调用API

URI

POST /v2/{project_id}/instances/{instance_id}/topics

表1 路径参数

参数

是否必选

参数类型

描述

project_id

String

参数解释

项目ID,获取方式请参见获取项目ID

约束限制

不涉及。

取值范围

不涉及。

默认取值

不涉及。

instance_id

String

参数解释

实例ID。获取方法如下:调用“查询所有实例列表”接口,从响应体中获取实例ID。

约束限制

不涉及。

取值范围

不涉及。

默认取值

不涉及。

请求参数

表2 请求Body参数

参数

是否必选

参数类型

描述

id

String

参数解释

Topic名称。

约束限制

长度为3-200,以字母开头且只支持大小写字母、中横线、下划线、点以及数字。

取值范围

不涉及

默认取值

不涉及。

replication

Integer

参数解释

副本数,配置数据的可靠性。

约束限制

副本数和代理数有关,如果有3个代理,最大副本数是3。

取值范围

不涉及

默认取值

不涉及。

sync_message_flush

Boolean

参数解释

是否使用同步落盘,同步落盘会导致性能降低。

约束限制

不涉及。

取值范围

  • true:同步落盘。

  • false:不使用同步落盘。

默认取值

false

partition

Integer

参数解释

Topic分区数,设置消费的并发数。

约束限制

不涉及。

取值范围

1-200。

默认取值

不涉及。

sync_replication

Boolean

参数解释

是否开启同步复制。

约束限制

不涉及。

取值范围

  • true:开启。开启后,客户端生产消息时相应的也要设置acks=-1,否则不生效。

  • false:不开启。

默认取值

false。

retention_time

Integer

参数解释

消息老化时间。

约束限制

不涉及。

取值范围

1-720

默认取值

72

topic_other_configs

Array of topic_other_configs objects

参数解释

Topic配置

topic_desc

String

参数解释

Topic描述。

约束限制

不涉及。

取值范围

0-200个字符。

默认取值

不涉及。

表3 topic_other_configs

参数

是否必选

参数类型

描述

name

String

参数解释

配置名称。

约束限制

不涉及。

取值范围

不涉及

默认取值

不涉及。

value

String

参数解释

配置值。

约束限制

不涉及。

取值范围

不涉及

默认取值

不涉及。

响应参数

状态码:200

表4 响应Body参数

参数

参数类型

描述

name

String

参数解释

Topic名称。

取值范围

不涉及

请求示例

创建一个Topic,Topic名称为test01。

POST https://{endpoint}/v2/{project_id}/instances/{instance_id}/topics

{
  "id" : "test01",
  "partition" : 3,
  "replication" : 3,
  "retention_time" : 72,
  "sync_message_flush" : false,
  "sync_replication" : "false",
  "topic_other_configs" : [ {
    "name" : "message.timestamp.type",
    "value" : "LogAppendTime"
  }, {
    "name" : "max.message.bytes",
    "value" : 10485760
  } ],
  "topic_desc" : ""
}

响应示例

状态码:200

创建成功,返回topic名称

{
  "name" : "test01"
}

SDK代码示例

SDK代码示例如下。

创建一个Topic,Topic名称为test01。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
package com.huaweicloud.sdk.test;

import com.huaweicloud.sdk.core.auth.ICredential;
import com.huaweicloud.sdk.core.auth.BasicCredentials;
import com.huaweicloud.sdk.core.exception.ConnectionException;
import com.huaweicloud.sdk.core.exception.RequestTimeoutException;
import com.huaweicloud.sdk.core.exception.ServiceResponseException;
import com.huaweicloud.sdk.kafka.v2.region.KafkaRegion;
import com.huaweicloud.sdk.kafka.v2.*;
import com.huaweicloud.sdk.kafka.v2.model.*;

import java.util.List;
import java.util.ArrayList;

public class CreateInstanceTopicSolution {

    public static void main(String[] args) {
        // The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security.
        // In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment
        String ak = System.getenv("CLOUD_SDK_AK");
        String sk = System.getenv("CLOUD_SDK_SK");
        String projectId = "{project_id}";

        ICredential auth = new BasicCredentials()
                .withProjectId(projectId)
                .withAk(ak)
                .withSk(sk);

        KafkaClient client = KafkaClient.newBuilder()
                .withCredential(auth)
                .withRegion(KafkaRegion.valueOf("<YOUR REGION>"))
                .build();
        CreateInstanceTopicRequest request = new CreateInstanceTopicRequest();
        request.withInstanceId("{instance_id}");
        CreateInstanceTopicReq body = new CreateInstanceTopicReq();
        List<CreateInstanceTopicReqTopicOtherConfigs> listbodyTopicOtherConfigs = new ArrayList<>();
        listbodyTopicOtherConfigs.add(
            new CreateInstanceTopicReqTopicOtherConfigs()
                .withName("message.timestamp.type")
                .withValue("LogAppendTime")
        );
        listbodyTopicOtherConfigs.add(
            new CreateInstanceTopicReqTopicOtherConfigs()
                .withName("max.message.bytes")
                .withValue("10485760")
        );
        body.withTopicDesc("");
        body.withTopicOtherConfigs(listbodyTopicOtherConfigs);
        body.withRetentionTime(72);
        body.withSyncReplication(false);
        body.withPartition(3);
        body.withSyncMessageFlush(false);
        body.withReplication(3);
        body.withId("test01");
        request.withBody(body);
        try {
            CreateInstanceTopicResponse response = client.createInstanceTopic(request);
            System.out.println(response.toString());
        } catch (ConnectionException e) {
            e.printStackTrace();
        } catch (RequestTimeoutException e) {
            e.printStackTrace();
        } catch (ServiceResponseException e) {
            e.printStackTrace();
            System.out.println(e.getHttpStatusCode());
            System.out.println(e.getRequestId());
            System.out.println(e.getErrorCode());
            System.out.println(e.getErrorMsg());
        }
    }
}

创建一个Topic,Topic名称为test01。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
# coding: utf-8

import os
from huaweicloudsdkcore.auth.credentials import BasicCredentials
from huaweicloudsdkkafka.v2.region.kafka_region import KafkaRegion
from huaweicloudsdkcore.exceptions import exceptions
from huaweicloudsdkkafka.v2 import *

if __name__ == "__main__":
    # The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security.
    # In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment
    ak = os.environ["CLOUD_SDK_AK"]
    sk = os.environ["CLOUD_SDK_SK"]
    projectId = "{project_id}"

    credentials = BasicCredentials(ak, sk, projectId)

    client = KafkaClient.new_builder() \
        .with_credentials(credentials) \
        .with_region(KafkaRegion.value_of("<YOUR REGION>")) \
        .build()

    try:
        request = CreateInstanceTopicRequest()
        request.instance_id = "{instance_id}"
        listTopicOtherConfigsbody = [
            CreateInstanceTopicReqTopicOtherConfigs(
                name="message.timestamp.type",
                value="LogAppendTime"
            ),
            CreateInstanceTopicReqTopicOtherConfigs(
                name="max.message.bytes",
                value="10485760"
            )
        ]
        request.body = CreateInstanceTopicReq(
            topic_desc="",
            topic_other_configs=listTopicOtherConfigsbody,
            retention_time=72,
            sync_replication=False,
            partition=3,
            sync_message_flush=False,
            replication=3,
            id="test01"
        )
        response = client.create_instance_topic(request)
        print(response)
    except exceptions.ClientRequestException as e:
        print(e.status_code)
        print(e.request_id)
        print(e.error_code)
        print(e.error_msg)

创建一个Topic,Topic名称为test01。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
package main

import (
	"fmt"
	"github.com/huaweicloud/huaweicloud-sdk-go-v3/core/auth/basic"
    kafka "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2"
	"github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2/model"
    region "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/kafka/v2/region"
)

func main() {
    // The AK and SK used for authentication are hard-coded or stored in plaintext, which has great security risks. It is recommended that the AK and SK be stored in ciphertext in configuration files or environment variables and decrypted during use to ensure security.
    // In this example, AK and SK are stored in environment variables for authentication. Before running this example, set environment variables CLOUD_SDK_AK and CLOUD_SDK_SK in the local environment
    ak := os.Getenv("CLOUD_SDK_AK")
    sk := os.Getenv("CLOUD_SDK_SK")
    projectId := "{project_id}"

    auth := basic.NewCredentialsBuilder().
        WithAk(ak).
        WithSk(sk).
        WithProjectId(projectId).
        Build()

    client := kafka.NewKafkaClient(
        kafka.KafkaClientBuilder().
            WithRegion(region.ValueOf("<YOUR REGION>")).
            WithCredential(auth).
            Build())

    request := &model.CreateInstanceTopicRequest{}
	request.InstanceId = "{instance_id}"
	nameTopicOtherConfigs:= "message.timestamp.type"
	valueTopicOtherConfigs:= "LogAppendTime"
	nameTopicOtherConfigs1:= "max.message.bytes"
	valueTopicOtherConfigs1:= "10485760"
	var listTopicOtherConfigsbody = []model.CreateInstanceTopicReqTopicOtherConfigs{
        {
            Name: &nameTopicOtherConfigs,
            Value: &valueTopicOtherConfigs,
        },
        {
            Name: &nameTopicOtherConfigs1,
            Value: &valueTopicOtherConfigs1,
        },
    }
	topicDescCreateInstanceTopicReq:= ""
	retentionTimeCreateInstanceTopicReq:= int32(72)
	syncReplicationCreateInstanceTopicReq:= false
	partitionCreateInstanceTopicReq:= int32(3)
	syncMessageFlushCreateInstanceTopicReq:= false
	replicationCreateInstanceTopicReq:= int32(3)
	request.Body = &model.CreateInstanceTopicReq{
		TopicDesc: &topicDescCreateInstanceTopicReq,
		TopicOtherConfigs: &listTopicOtherConfigsbody,
		RetentionTime: &retentionTimeCreateInstanceTopicReq,
		SyncReplication: &syncReplicationCreateInstanceTopicReq,
		Partition: &partitionCreateInstanceTopicReq,
		SyncMessageFlush: &syncMessageFlushCreateInstanceTopicReq,
		Replication: &replicationCreateInstanceTopicReq,
		Id: "test01",
	}
	response, err := client.CreateInstanceTopic(request)
	if err == nil {
        fmt.Printf("%+v\n", response)
    } else {
        fmt.Println(err)
    }
}

更多编程语言的SDK代码示例,请参见API Explorer的代码示例页签,可生成自动对应的SDK代码示例。

状态码

状态码

描述

200

创建成功,返回topic名称

错误码

请参见错误码