काफ्का 0.8, क्या जावा कोड का उपयोग करके विभाजन और प्रतिकृति के साथ विषय बनाना संभव है?

Question

काफ्का में 0.8beta नीचे कोई विषय की तरह एक आदेश का उपयोग कर के रूप में उल्लेख here

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

उपरोक्त आदेश 3 विभाजन और 2 विभाजन प्रति प्रतिकृतियां के साथ "परीक्षण" नाम के एक विषय पैदा करेगा बनाया जा सकता है।

क्या मैं जावा का उपयोग कर वही काम कर सकता हूं?

अब तक मैं क्या पाया जावा उपयोग कर रहा है जैसा कि नीचे

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

इस विभाजन की संख्या के साथ "mytopic" निर्दिष्ट "num.partitions" का उपयोग कर नाम के एक विषय बनाएगा देखा एक निर्माता बना सकते हैं विशेषता और उत्पादन शुरू करते हैं।

लेकिन क्या विभाजन और प्रतिकृति को परिभाषित करने का कोई तरीका है? मुझे ऐसा कोई उदाहरण नहीं मिला। यदि हम ऐसा नहीं कर सकते हैं तो इसका मतलब है कि हमें हमेशा विभाजन और प्रतिकृति (हमारी आवश्यकता के अनुसार) के साथ विषय बनाने की आवश्यकता है और फिर उस विषय के भीतर संदेश का उत्पादन करने के लिए निर्माता का उपयोग करें। उदाहरण के लिए यदि मैं "mytopic" को उसी तरह से बनाना चाहता हूं, लेकिन विभाजन की विभिन्न संख्या (num.partitions विशेषता को ओवरराइड करना) के साथ यह संभव होगा?

Answer 1

नोट: मेरा जवाब को शामिल किया गया काफ्का 0.8.1+, यानि कि नवीनतम स्थिर अप्रैल 2014

हाँ के रूप में उपलब्ध संस्करण, आप काफ्का एपीआई के माध्यम से प्रोग्राम के एक विषय बना सकते हैं। और हां, आप विभाजन के वांछित संख्या के साथ-साथ विषय के लिए प्रतिकृति कारक निर्दिष्ट कर सकते हैं।

ध्यान दें कि हाल ही में जारी किए गए काफ्का 0.8.1+ काफ्का 0.8.0 (जिसका उपयोग बीक द्वारा उनके लिंक किए गए उत्तर में किया गया था) से थोड़ा अलग एपीआई प्रदान करता है। मैंने How Can we create a topic in Kafka from the IDE using API प्रश्न के उत्तर में code example to create a topic in Kafka 0.8.1+ जोड़ा जो कि बीक ऊपर उल्लिखित थे।

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