接口调用示例

更新时间：2026-09-09

集群管理

createCluster 创建集群

功能说明

创建 Kafka 集群。CreateClusterRequest 只强制校验 name，mode、type 与 provisioned 内部字段的合法组合由服务端校验。创建预置型集群时应完整填写 Provisioned，其中基本类型字段需显式赋值。

创建集群会产生计费，预付费场景下 Billing.isAutoPay 默认为 true，会直接扣费。

请求示例

                Java
                
            

                package com.example.kafka;

import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.Map;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.cluster.Authentication;
import com.baidubce.services.kafka.model.cluster.Billing;
import com.baidubce.services.kafka.model.cluster.ConfigMeta;
import com.baidubce.services.kafka.model.cluster.CreateClusterRequest;
import com.baidubce.services.kafka.model.cluster.CreateClusterResponse;
import com.baidubce.services.kafka.model.cluster.Mode;
import com.baidubce.services.kafka.model.cluster.Provisioned;
import com.baidubce.services.kafka.model.cluster.StorageMeta;
import com.baidubce.services.kafka.model.cluster.StorageType;
import com.baidubce.services.kafka.model.cluster.Tag;
import com.baidubce.services.kafka.model.cluster.Type;

public class CreateClusterDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            CreateClusterRequest request = new CreateClusterRequest();
            request.setName("kafka-demo");
            request.setMode(Mode.HP);
            request.setType(Type.PROVISIONED);
            request.setTags(Arrays.asList(
                    Tag.builder().tagKey("KAFKA-Cluster").tagValue("prod").build()));

            Provisioned provisioned = new Provisioned();
            provisioned.setKafkaVersion("2.7.2");
            provisioned.setNodeType("kafka.g4.c2m8");
            provisioned.setNumberOfBrokerNodes(3);
            provisioned.setStorageMeta(StorageMeta.builder()
                    .storageType(StorageType.ENHANCED_SSD_PL1)
                    .storageSize(100)
                    .numberOfDisk(1)
                    .build());
            provisioned.setVpcId("{{VPC ID}}");
            provisioned.setSubnetIds(Arrays.asList("{{子网 ID}}"));
            provisioned.setSecurityGroupIds(Arrays.asList("{{安全组 ID}}"));
            provisioned.setPublicIpEnabled(false);
            provisioned.setIntranetIpEnabled(false);
            provisioned.setAclEnabled(true);
            provisioned.setDeploySetEnabled(true);
            provisioned.setAuthentications(Arrays.asList(
                    Authentication.builder().mode("SASL_SCRAM").build()));
            provisioned.setBilling(Billing.builder()
                    .payment("Prepaid")
                    .timeLength(1)
                    .timeUnit("month")
                    .autoRenewEnabled(false)
                    .isAutoPay(true)
                    .build());

            Map<String, String> configContext = new LinkedHashMap<String, String>();
            configContext.put("auto.create.topics.enable", "true");
            configContext.put("log.retention.hours", "168");
            provisioned.setConfigMeta(ConfigMeta.builder().context(configContext).build());

            request.setProvisioned(provisioned);

            CreateClusterResponse response = client.createCluster(request);
            System.out.println("clusterId=" + response.getClusterId());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

CreateClusterResponse 字段结构：

Plain Text

1clusterId : String   集群 ID

集群创建是异步过程，返回 clusterId 只表示创建请求已受理，需继续查询集群状态确认就绪。

相关 API

创建集群

getClusterDetail 查询集群详情

功能说明

按 clusterId 查询集群详情。响应中的 ClusterDetail.provisioned 包含规格、网络、存储、认证和维护窗口等完整配置。

查询已删除集群请使用 getClusterDeletion，其请求类型为 GetClusterDeletionRequest，但响应复用 GetClusterDetailResponse。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.cluster.ClusterDetail;
import com.baidubce.services.kafka.model.cluster.GetClusterDetailRequest;
import com.baidubce.services.kafka.model.cluster.GetClusterDetailResponse;

public class GetClusterDetailDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            GetClusterDetailRequest request = new GetClusterDetailRequest();
            request.setClusterId("{{集群 ID}}");

            GetClusterDetailResponse response = client.getClusterDetail(request);
            ClusterDetail detail = response.getCluster();
            if (detail != null) {
                System.out.printf("name=%s state=%s region=%s%n",
                        detail.getName(), detail.getState(), detail.getRegion());
            }
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

GetClusterDetailResponse.cluster（ClusterDetail）字段结构：

Plain Text

1clusterId   : String        集群 ID
2clusterSid  : String        集群短 ID
3name        : String        集群名称
4region      : String        地域
5type         : String       集群类型
6mode         : String       集群模式
7vpcMode      : String       VPC 资源管理模式
8state        : String       集群状态
9provisioned  : Provisioned  预置模式配置
10tags         : List<Tag>    标签列表
11createTime   : String       创建时间
12deleteTime   : String       删除时间

相关 API

查询集群详情

stopCluster 停止集群

功能说明

停止运行中的集群。停止后集群不再提供服务，客户端连接会中断，属于影响业务的操作，执行前需确认目标集群。

接口返回 actionId，需通过任务接口确认最终状态。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.cluster.StopClusterRequest;
import com.baidubce.services.kafka.model.cluster.StopClusterResponse;

public class StopClusterDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            StopClusterRequest request = new StopClusterRequest();
            request.setClusterId("{{集群 ID}}");

            StopClusterResponse response = client.stopCluster(request);
            System.out.printf("clusterId=%s actionId=%s%n",
                    response.getClusterId(), response.getActionId());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

StopClusterResponse 字段结构：

Plain Text

1clusterId : String   集群 ID
2actionId  : String   任务 ID，用于后续查询任务状态

相关 API

停止集群

主题管理

createTopic 创建主题

功能说明

在指定集群下创建主题。partitionNum 与 replicationFactor 是 int 基本类型，不赋值会发送 0，必须显式设置。

可用的主题级参数及取值范围通过 listTopicConfigOptions() 查询，该方法不接受请求对象。

请求示例

                Java
                
            

                package com.example.kafka;

import java.util.LinkedHashMap;
import java.util.Map;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.topic.CreateTopicRequest;
import com.baidubce.services.kafka.model.topic.CreateTopicResponse;

public class CreateTopicDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            CreateTopicRequest request = new CreateTopicRequest();
            request.setClusterId("{{集群 ID}}");
            request.setTopicName("{{主题名称}}");
            request.setPartitionNum(3);
            request.setReplicationFactor(3);

            Map<String, String> otherConfigs = new LinkedHashMap<String, String>();
            otherConfigs.put("cleanup.policy", "delete");
            otherConfigs.put("retention.ms", "604800000");
            request.setOtherConfigs(otherConfigs);

            CreateTopicResponse response = client.createTopic(request);
            System.out.println("topicName=" + response.getTopicName());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

CreateTopicResponse 字段结构：

Plain Text

1topicName : String   主题名称

相关 API

创建主题

updateTopic 变更主题

功能说明

变更主题分区数或主题级配置。partitionNum 与 otherConfigs 至少设置一项，否则抛出 IllegalArgumentException。分区数只能增加，不能减少。

UpdateTopicRequest.partitionNum 的字段类型是 String，同时提供 setPartitionNum(int) 重载完成转换，推荐使用 int 重载避免手工拼字符串。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.topic.UpdateTopicRequest;
import com.baidubce.services.kafka.model.topic.UpdateTopicResponse;

public class UpdateTopicDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            UpdateTopicRequest request = new UpdateTopicRequest();
            request.setClusterId("{{集群 ID}}");
            request.setTopicName("{{主题名称}}");
            request.setPartitionNum(6);

            UpdateTopicResponse response = client.updateTopic(request);
            System.out.println("topicName=" + response.getTopicName());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

UpdateTopicResponse 字段结构：

Plain Text

1topicName : String   主题名称

相关 API

变更主题

listTopic 查询主题列表

功能说明

查询集群下的主题列表。topicName 可选，按前缀匹配过滤。该接口不分页，返回集群下符合条件的全部主题。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.topic.ListTopicRequest;
import com.baidubce.services.kafka.model.topic.ListTopicResponse;
import com.baidubce.services.kafka.model.topic.Topic;

public class ListTopicDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            ListTopicRequest request = new ListTopicRequest();
            request.setClusterId("{{集群 ID}}");
            request.setTopicName("order-");

            ListTopicResponse response = client.listTopic(request);
            for (Topic topic : response.getTopics()) {
                System.out.printf("topicName=%s partitionNum=%d replicaNum=%d%n",
                        topic.getTopicName(), topic.getPartitionNum(), topic.getReplicaNum());
            }
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

ListTopicResponse.topics 元素（Topic）字段结构：

Plain Text

1topicName    : String    主题名称
2createTime   : String    创建时间
3readOnly     : Boolean   是否只读
4partitionNum : Integer   分区数
5replicaNum   : Integer   副本数

相关 API

查询主题列表

消费组管理

listConsumerGroup 查询消费组列表

功能说明

查询集群下的消费组列表。groupName 可选，用于过滤。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.consumer.Group;
import com.baidubce.services.kafka.model.consumer.ListConsumerGroupRequest;
import com.baidubce.services.kafka.model.consumer.ListConsumerGroupResponse;

public class ListConsumerGroupDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            ListConsumerGroupRequest request = new ListConsumerGroupRequest();
            request.setClusterId("{{集群 ID}}");

            ListConsumerGroupResponse response = client.listConsumerGroup(request);
            for (Group group : response.getGroups()) {
                System.out.printf("groupName=%s state=%s coordinator=%s%n",
                        group.getGroupName(), group.getState(), group.getGroupCoordinatorId());
            }
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

ListConsumerGroupResponse.groups 元素（Group）字段结构：

Plain Text

1groupName          : String    消费组名称
2updateTime         : String    更新时间
3state              : String    消费组状态
4groupCoordinatorId : Integer   Coordinator 所在 Broker ID

相关 API

查询消费组列表

resetConsumerGroup 重置消费组位点

功能说明

重置消费组在指定主题分区上的消费位点。topicName、partitions、resetStrategy 都是必填，partitions 不能为空列表。

重置位点会直接改变消费进度，可能导致消息重复消费或跳过，执行前必须确认目标消费组、主题和分区。建议在消费组无活跃成员时执行。

请求示例

                Java
                
            

                package com.example.kafka;

import java.util.Arrays;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.consumer.ResetConsumerGroupRequest;
import com.baidubce.services.kafka.model.consumer.ResetConsumerGroupResponse;

public class ResetConsumerGroupDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            ResetConsumerGroupRequest request = new ResetConsumerGroupRequest();
            request.setClusterId("{{集群 ID}}");
            request.setGroupName("{{消费组名称}}");
            request.setTopicName("{{主题名称}}");
            request.setPartitions(Arrays.asList(0, 1, 2));
            request.setResetStrategy("{{重置策略}}");
            request.setResetValue("{{重置目标值}}");

            ResetConsumerGroupResponse response = client.resetConsumerGroup(request);
            System.out.println("groupName=" + response.getGroupName());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

ResetConsumerGroupResponse 字段结构：

Plain Text

1groupName : String   消费组名称

相关 API

重置消费位点

用户管理

createUser 创建用户

功能说明

在集群下创建 SASL 用户。SDK 内部使用凭证 SK 的前 16 个字符作为密钥，以 AES/ECB/PKCS5Padding 加密密码后以大写十六进制字符串发送，调用方传入原始密码即可。

saslMechanisms 可选，取值示例为 SCRAM-SHA-256、SCRAM-SHA-512、PLAIN。

加密会直接写回请求对象的 password 字段。 同一个 CreateUserRequest 实例不能重复提交，否则第二次调用会对已加密的值再次加密。每次调用请新建请求对象。同样的约束适用于 ResetUserPasswordRequest。

不要记录原始密码、加密后的密码或完整请求体。

请求示例

                Java
                
            

                package com.example.kafka;

import java.util.Arrays;
import java.util.HashSet;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.user.CreateUserRequest;
import com.baidubce.services.kafka.model.user.CreateUserResponse;

public class CreateUserDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            CreateUserRequest request = new CreateUserRequest();
            request.setClusterId("{{集群 ID}}");
            request.setUsername("{{用户名}}");
            request.setPassword(System.getenv("KAFKA_USER_PASSWORD"));
            request.setSaslMechanisms(new HashSet<String>(
                    Arrays.asList("SCRAM-SHA-256", "SCRAM-SHA-512")));

            CreateUserResponse response = client.createUser(request);
            System.out.println("username=" + response.getUsername());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

CreateUserResponse 字段结构：

Plain Text

1username : String   用户名

相关 API

创建用户

listUsers 查询用户列表

功能说明

查询集群下的用户列表。listUsers(String clusterId) 是已标注 @Deprecated 的便捷重载，内部委托到 listUsers(ListUsersRequest)，新代码应直接使用请求对象。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.user.ListUserResponse;
import com.baidubce.services.kafka.model.user.ListUsersRequest;
import com.baidubce.services.kafka.model.user.User;

public class ListUsersDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            ListUsersRequest request = new ListUsersRequest();
            request.setClusterId("{{集群 ID}}");

            ListUserResponse response = client.listUsers(request);
            for (User user : response.getUsers()) {
                System.out.printf("username=%s createTime=%s mechanisms=%s%n",
                        user.getUsername(), user.getCreateTime(), user.getSaslMechanisms());
            }
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

ListUserResponse.users 元素（User）字段结构：

Plain Text

1username       : String        用户名
2createTime     : String        创建时间
3saslMechanisms : Set<String>   已启用的 SASL 机制

相关 API

查询用户列表

权限管理

createAcl 创建权限

功能说明

为指定用户创建 ACL 授权。username、patternType、resourceType、resourceName 必填，operations 不能为空列表。

注意 createAcl 使用 operations（List<String>，可一次授予多个操作），而 deleteAcl 使用 operation（单个 String）。

请求示例

                Java
                
            

                package com.example.kafka;

import java.util.Arrays;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.acl.CreateAclRequest;
import com.baidubce.services.kafka.model.acl.CreateAclResponse;

public class CreateAclDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            CreateAclRequest request = new CreateAclRequest();
            request.setClusterId("{{集群 ID}}");
            request.setUsername("{{用户名}}");
            request.setPatternType("{{匹配模式}}");
            request.setResourceType("{{资源类型}}");
            request.setResourceName("{{资源名称}}");
            request.setOperations(Arrays.asList("{{操作类型}}"));

            CreateAclResponse response = client.createAcl(request);
            System.out.println("username=" + response.getUsername());
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

CreateAclResponse 字段结构：

Plain Text

1username : String   用户名

相关 API

创建权限

listAcls 查询权限列表

功能说明

查询集群下的 ACL 列表，支持按 username、patternType、resourceType、resourceName 过滤。

listAcls(String clusterId) 与 listAcls(ListAclRequest) 两个重载都可用，后者支持过滤条件。

请求示例

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.acl.Acl;
import com.baidubce.services.kafka.model.acl.ListAclRequest;
import com.baidubce.services.kafka.model.acl.ListAclResponse;

public class ListAclsDemo {

    public static void main(String[] args) {
        KafkaClient client = KafkaClientFactory.create();
        try {
            ListAclRequest request = ListAclRequest.builder()
                    .clusterId("{{集群 ID}}")
                    .username("{{用户名}}")
                    .build();

            ListAclResponse response = client.listAcls(request);
            for (Acl acl : response.getAcls()) {
                System.out.printf("username=%s resourceType=%s resourceName=%s operation=%s%n",
                        acl.getUsername(), acl.getResourceType(),
                        acl.getResourceName(), acl.getOperation());
            }
        } finally {
            client.shutdown();
        }
    }
}
            

返回示例

ListAclResponse.acls 元素（Acl）字段结构：

Plain Text

1username     : String   用户名
2patternType  : String   匹配模式
3resourceType : String   资源类型
4resourceName : String   资源名称
5operation    : String   操作类型

相关 API

查询权限列表

任务管理

资源变更、主题重新分区等操作可能异步执行。SDK 调用成功仅表示任务创建成功，不代表操作已经完成。

请继续查询任务状态，直至任务进入成功或失败状态。多数变更接口的响应包含 actionId 字段，即任务 ID。

                Java
                
            

                package com.example.kafka;

import com.baidubce.services.kafka.KafkaClient;
import com.baidubce.services.kafka.model.cluster.IncreaseBrokerCountRequest;
import com.baidubce.services.kafka.model.cluster.IncreaseBrokerCountResponse;
import com.baidubce.services.kafka.model.job.GetJobDetailRequest;
import com.baidubce.services.kafka.model.job.GetJobDetailResponse;
import com.baidubce.services.kafka.model.job.Job;
import com.baidubce.services.kafka.model.job.Operation;

public class AsyncJobDemo {

    public static void main(String[] args) throws InterruptedException {
        KafkaClient client = KafkaClientFactory.create();
        try {
            String clusterId = "{{集群 ID}}";

            // 1. 提交异步变更，取回 actionId
            IncreaseBrokerCountRequest increaseRequest = new IncreaseBrokerCountRequest();
            increaseRequest.setClusterId(clusterId);
            increaseRequest.setNumberOfBrokerNodes(6);

            IncreaseBrokerCountResponse increaseResponse =
                    client.increaseBrokerCount(increaseRequest);
            String actionId = increaseResponse.getActionId();
            System.out.println("actionId=" + actionId);

            // 2. 轮询任务状态，直至进入终态
            while (true) {
                GetJobDetailRequest jobRequest = new GetJobDetailRequest();
                jobRequest.setClusterId(clusterId);
                jobRequest.setActionId(actionId);

                GetJobDetailResponse jobResponse = client.getJob(jobRequest);
                Job job = jobResponse.getJob();
                System.out.println("status=" + job.getStatus());

                if (job.getOperations() != null) {
                    for (Operation operation : job.getOperations()) {
                        System.out.printf("  operationId=%s type=%s state=%s process=%s%n",
                                operation.getOperationId(), operation.getType(),
                                operation.getState(), operation.getProcess());
                    }
                }

                if (isFinished(job.getStatus())) {
                    break;
                }
                Thread.sleep(10 * 1000L);
            }
        } finally {
            client.shutdown();
        }
    }

    private static boolean isFinished(String status) {
        // 终态取值请以服务端返回为准
        return "{{成功状态}}".equals(status) || "{{失败状态}}".equals(status);
    }
}
            

getOperation 返回 OperationDetail，包含 process、schedule、groups、sourceContext、targetContext 等更细粒度的进度信息。startJob、cancelJob、suspendJob、resumeJob 用于控制任务执行。

轮询间隔建议不小于 10 秒，避免高频调用触发限流。

百度智能云

消息服务 for Kafka

消息服务 for Kafka

接口调用示例

集群管理

createCluster 创建集群

getClusterDetail 查询集群详情

stopCluster 停止集群

主题管理

createTopic 创建主题

updateTopic 变更主题

listTopic 查询主题列表

消费组管理

listConsumerGroup 查询消费组列表

resetConsumerGroup 重置消费组位点

用户管理

createUser 创建用户

listUsers 查询用户列表

权限管理

createAcl 创建权限

listAcls 查询权限列表

任务管理