Kafka 0.8, có thể tạo chủ đề với phân vùng và sao chép bằng cách sử dụng mã java không?

Question

Trong Kafka 0.8beta một chủ đề có thể được tạo ra bằng cách sử dụng lệnh như dưới đây như đã đề cập here

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

lệnh trên sẽ tạo ra một chủ đề có tên là "thử nghiệm" với 3 phân vùng và 2 bản sao cho mỗi phân vùng.

Tôi có thể làm điều tương tự bằng cách sử dụng Java không?

Cho đến nay những gì tôi thấy đang sử dụng Java chúng ta có thể tạo ra một nhà sản xuất như bên dưới

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

này sẽ tạo ra một chủ đề có tên là "mytopic" với số lượng phân vùng quy định sử dụng "num.partitions" thuộc tính và bắt đầu sản xuất.

Nhưng có cách nào để xác định phân vùng và sao chép không? Tôi không thể tìm thấy bất kỳ ví dụ nào như vậy. Nếu chúng ta không thể làm điều đó có nghĩa là chúng ta luôn cần tạo chủ đề với phân vùng và sao chép (theo yêu cầu của chúng ta) trước và sau đó sử dụng nhà sản xuất để tạo ra thông điệp trong chủ đề đó. Ví dụ, nó sẽ có thể nếu tôi muốn tạo ra "mytopic" theo cùng một cách nhưng với số lượng phân vùng khác nhau (ghi đè thuộc tính num.partitions)?

Answer 1

Lưu ý: Câu trả lời của tôi bao gồm Kafka 0.8.1+, ví dụ phiên bản ổn định mới nhất có sẵn như tháng Tư năm 2014.

Vâng, bạn có thể tạo một chủ đề programatically qua API Kafka. Và có, bạn có thể chỉ định số lượng phân vùng mong muốn cũng như nhân tố sao chép cho chủ đề.

Lưu ý rằng Kafka 0.8.1+ mới phát hành cung cấp một API hơi khác so với Kafka 0.8.0 (được sử dụng bởi Biks trong câu trả lời được liên kết của anh ấy). Tôi đã thêm một số code example to create a topic in Kafka 0.8.1+ vào câu trả lời của mình cho câu hỏi How Can we create a topic in Kafka from the IDE using API mà Biks đang đề cập ở trên.

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