Kafka 0.8, czy można utworzyć temat z partycją i replikacją za pomocą kodu java?

Question

W Kafki 0.8beta tematem mogą być tworzone za pomocą polecenia jak poniżej wymienione here

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

powyższe polecenie stworzy temat o nazwie „test” z 3 przegrodami i 2 replik na partycji.

Czy mogę zrobić to samo za pomocą Java?

tej pory co znalazłem się przy użyciu języka Java możemy stworzyć producent jak widać poniżej

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

Stworzy to wątek o nazwie „mytopic” z liczbą partycji określony za pomocą „num.partitions” atrybut i zacznij produkcję.

Ale czy istnieje sposób zdefiniowania partycji i replikacji również? Nie mogłem znaleźć żadnego takiego przykładu. Jeśli nie możemy tego zrobić, oznacza to, że zawsze musimy utworzyć temat z partycjami i replikacją (zgodnie z naszym wymaganiem), a następnie użyć producenta do wytworzenia wiadomości w tym temacie. Na przykład czy będzie możliwe, jeśli chcę utworzyć "mytopic" w ten sam sposób, ale z inną liczbą partycji (przesłonięcie atrybutu num.partitions)?

Answer 1

Uwaga: Moja odpowiedź obejmuje Kafka 0.8.1+, czyli najnowsza stabilna wersja dostępna od kwietnia 2014 r

Tak, można utworzyć wątek programowo za pośrednictwem interfejsu API Kafki. I tak, można określić żądaną liczbę partycji, a także współczynnik replikacji dla tematu.

Należy zauważyć, że ostatnio wydany Kafka 0.8.1+ zapewnia nieco inne API niż Kafka 0.8.0 (który był używany przez Biks w jego połączonej odpowiedzi). Dodałem code example to create a topic in Kafka 0.8.1+ do mojej odpowiedzi na pytanie How Can we create a topic in Kafka from the IDE using API, które Biks miał na myśli powyżej.

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