是的，有可能。您必须在中访问
AdminUtilsscala类
kafka_2.11-0.9.0.1.jar以添加分区。
AdminUtils支持主题中的分区数只能增加。您可能需要
kafka_2.11-0.9.0.1.jar，
zk-client-0.8.jar，
scala-library-2.11.8.jar和
scala-parser-combinators_2.11-1.0.4.jar在类路径罐子。
以下代码的某些部分是从kafka-cloudera示例中借用/启发的。
package org.apache.kafka.examples;import java.io.Closeable;import org.I0Itec.zkclient.ZkClient;import org.apache.logging.log4j.LogManager;import org.apache.logging.log4j.Logger;import kafka.admin.AdminOperationException;import kafka.admin.AdminUtils;import kafka.admin.RackAwareMode.Enforced$;import kafka.utils.ZKStringSerializer$;import kafka.utils.ZkUtils;public class Test { static final Logger logger = LogManager.getLogger(); public Test() { // TODO Auto-generated constructor stub } public static void addPartitions(String zkServers, String topic, int partitions) { try (AutoZkClient zkClient = new AutoZkClient(zkServers)) { ZkUtils zkUtils = ZkUtils.apply(zkClient, false); if (AdminUtils.topicExists(zkUtils, topic)) { logger.info("Altering topic {}", topic); try { AdminUtils.addPartitions(zkUtils, topic, partitions, "", true, Enforced$.MODULE$); logger.info("Topic {} altered with partitions : {}", topic, partitions); } catch (AdminOperationException aoe) { logger.info("Error while altering partitions for topic : {}", topic, aoe); } } else { logger.info("Topic {} doesn't exists", topic); } } } // Just exists for Closeable convenience private static final class AutoZkClient extends ZkClient implements Closeable { static int sessionTimeout = 30_000; static int connectionTimeout = 6_000; AutoZkClient(String zkServers) { super(zkServers, sessionTimeout, connectionTimeout, ZKStringSerializer$.MODULE$); } } public static void main(String[] args) { addPartitions("localhost:2181", "hello", 20); }}
