Kafka: change the number of sections for a specific topic using java

Question

I am new to Kafka and working with the new KafkaProducer and KafkaConsumer, version: 0.9.0.1

Is there a way in java to change or update the number of sections for a particular topic after its creation.

I do not use zookeeper to create themes. My KafkaProducer automatically creates topics when I receive a publication request.

I can also provide more details if this is not enough.

Answer 1

Yes it is possible. You must access the AdminUtils scala class in kafka_2.11-0.9.0.1.jar to add sections.

AdminUtils supports the number of sections in a topic just to increase. You may need banks kafka_2.11-0.9.0.1.jar , zk-client-0.8.jar , scala-library-2.11.8.jar and scala-parser-combinators_2.11-1.0.4.jar in your class path .

Parts of the code below are borrowed / inspired by kafka-cloudera examples.

 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); } }