Kafka 0.8, java kodu kullanarak bölüm ve çoğaltma ile konu oluşturmak mümkün mü? Kafka yılında

Question

bir konu here

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

Yukarıdaki komut 3 bölümleri ve bölüm başına 2 kopyaları ile "test" adında bir konu yaratacaktır belirtildiği gibi aşağıdaki gibi bir komut kullanılarak oluşturulabilir 0.8beta.

Java ile aynı şeyi yapabilir miyim?

Şimdiye kadar

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

Bu "num.partitions" kullanılarak belirtilen bölümünün numarası ile "mytopic" adlı bir konu yaratacaktır

aşağıda görüldüğü gibi bir yapımcı oluşturabilir Java kullanıyor ne buldum nitelik ve üretmeye başlar.

Ama bölüm ve çoğaltmayı da tanımlamanın bir yolu var mı? Böyle bir örnek bulamadım. Bunu yapamazsak, her zaman önce bölümler ve çoğaltma (gereğimize göre) içeren bir konu yaratmamız ve daha sonra bu konuda mesaj üretmek için üreticiyi kullanmamız gerektiği anlamına gelir. Örneğin, "mytopic" i aynı şekilde ama farklı sayıda bölümle (num.partitions özniteliğini geçersiz kılarak) oluşturmak istersem mümkün olabilir mi?

Answer 1

Not: Benim cevabım Kafka 0.8.1+,

Evet Nisan 2014 tarihinden itibaren geçerli yani en son kararlı sürümü kapsar sen Kafka API üzerinden programlama yoluyla konu oluşturabilir. Ve evet, istediğiniz bölüm sayısını ve konu için çoğaltma faktörünü belirtebilirsiniz.

Yakın zamanda yayınlanan Kafka 0.8.1+ sürümünün Kafka 0.8.0'dan biraz daha farklı bir API oluşturduğunu unutmayın (Biks tarafından bağlı cevabında kullanıldı). Bikes'in yukarıda bahsettiği How Can we create a topic in Kafka from the IDE using API numaralı soruya verdiğim yanıt için code example to create a topic in Kafka 0.8.1+ ekledim. Bir konu oluşturulabilir `kafka.javaapi.producer.Producer` API ama bölüm sayısını kullanarak, bugüne kadar is buldum Ne

Answer 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();`