尝试用Java代码创建Kafka Partition失败，求实现实例指导

嘿，我懂你在用Java代码操作Kafka分区时碰壁的感受——其实Kafka的分区是和Topic绑定在一起的，没法单独“创建分区”，只能在创建Topic时指定分区数量，或者给已存在的Topic扩容增加分区。下面给你两个场景的完整可运行代码实例，用的是Kafka官方推荐的AdminClient API，这也是目前最稳定的方式：

首先得确保你的项目里引入了Kafka客户端依赖，比如Maven的pom.xml可以加这段：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.1</version> <!-- 替换成你实际使用的Kafka版本 -->
</dependency>

然后是Java代码示例：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CreateKafkaTopicWithPartitions {
    public static void main(String[] args) {
        // 配置AdminClient连接参数
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 替换成你的Kafka集群地址

        // 用try-with-resources自动管理AdminClient资源
        try (AdminClient adminClient = AdminClient.create(props)) {
            // 定义Topic：名称、分区数、副本因子（副本数不能超过集群broker数量）
            NewTopic newTopic = new NewTopic("my-test-topic", 5, (short) 1);
            // 发送创建请求并等待结果
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();
            System.out.println("Topic创建成功，包含5个分区！");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            System.err.println("创建Topic时出错：" + e.getMessage());
        }
    }
}

如果你的Topic已经存在，想要扩容分区数，可以用下面的代码：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreatePartitionsResult;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class AddPartitionsToKafkaTopic {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 定义要扩容的Topic和目标分区数（必须大于当前分区数）
            Map<String, Integer> topicPartitionCounts = new HashMap<>();
            topicPartitionCounts.put("my-test-topic", 8);

            // 发送扩容请求并等待完成
            CreatePartitionsResult result = adminClient.createPartitions(topicPartitionCounts);
            result.all().get();
            System.out.println("Topic分区扩容成功，现在有8个分区！");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
            System.err.println("扩容分区时出错：" + e.getMessage());
        }
    }
}

重要注意事项 #

• Kafka不支持减少分区数，只能增加，所以设置目标分区数时一定要比当前值大
• 确保你的Kafka集群配置允许手动修改Topic（如果是自动创建模式，auto.create.topics.enable需设为true，但手动创建/修改更可控）
• 运行代码的机器要能访问Kafka broker端口，且拥有对应的操作权限（创建/修改Topic的权限）

内容的提问来源于stack exchange，提问作者Sanjiv