Kafka 0.8, is it possible to create topic with par

问题:

In Kafka 0.8beta a topic can be created using a command like below as mentioned here

    bin/kafka-create-topic.sh --zookeeper localhost:2181 --replica 2 --partition 3 --topic test

the above command will create a topic named "test" with 3 partitions and 2 replicas per partition.

Can I do the same thing using Java ?

So far what I found is using Java we can create a producer as seen below

    Producer<String, String> producer = new Producer<String, String>(config);
    producer.send(new KeyedMessage<String, String>("mytopic", msg));

This will create a topic named "mytopic" with the number of partition specified using the "num.partitions" attribute and start producing.

But is there a way to define the partition and replication also ? I couldn't find any such example. If we can't then does that mean we always need to create topic with partitions and replication (as per our requirement) before and then use the producer to produce message within that topic. For example will it be possible if I want to create the "mytopic" the same way but with different number of partition (overriding the num.partitions attribute) ?

回答1:

Note: My answer covers Kafka 0.8.1+, i.e. the latest stable version available as of April 2014.

Yes, you can create a topic programatically via the Kafka API. And yes, you can specify the desired number of partitions as well as the replication factor for the topic.

Note that the recently released Kafka 0.8.1+ provides a slightly different API than Kafka 0.8.0 (which was used by Biks in his linked reply). I added a code example to create a topic in Kafka 0.8.1+ to my reply to the question How Can we create a topic in Kafka from the IDE using API that Biks was referring to above.

回答2:

`

import kafka.admin.AdminUtils;
import kafka.cluster.Broker;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;

String zkConnect = "localhost:2181";
ZkClient zkClient = new ZkClient(zkConnect, 10 * 1000, 8 * 1000, ZKStringSerializer$.MODULE$);
ZkUtils zkUtils = new ZkUtils(zkClient, new ZkConnection(zkConnect), false);
    Properties pop = new Properties();
    AdminUtils.createTopic(zkUtils, topic.getTopicName(), topic.getPartitionCount(), topic.getReplicationFactor(),
            pop);
    zkClient.close();`