我是Kafka的新手,正在使用新的KafkaProducer和KafkaConsumer,版本:0.9.0.1
在创建特定主题之后,java中是否有任何方法可以更改/更新特定主题的分区数。
我没有使用Zookeeper创建主题。当发布请求到达时,我的KafkaProducer会自动创建主题。
如果还不够,我还可以提供更多详细信息
是的,有可能。您必须在中访问AdminUtilsscala类kafka_2.11-0.9.0.1.jar以添加分区。
AdminUtils
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在类路径罐子。
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); } }