Kafka 0.8, можно ли создать тему с разделом и репликацией с помощью java-кода?

В Kafka 0.8beta можно создать тему, используя следующую команду: здесь

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

приведенная выше команда создаст тему с именем "test" с 3 разделами и 2 репликами на раздел.

Могу ли я сделать то же самое с помощью Java?

Пока что я нашел использование Java, мы можем создать производителя, как показано ниже.

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

Это создаст тему с именем "mytopic" с номером раздела, указанным с использованием атрибута "num.partitions" и начнет создавать.

Но есть ли способ определить раздел и репликацию? Я не мог найти такого примера. Если мы не можем, значит, это означает, что нам всегда нужно создавать тему с разделами и репликацией (в соответствии с нашим требованием) до и затем использовать продюсер для создания сообщения в этой теме. Например, возможно ли, если я хочу создать "mytopic" одинаково, но с разным количеством разделов (переопределяя атрибут num.partitions)?

Ответ 1

Примечание. Мой ответ охватывает Kafka 0.8.1+, т.е. последнюю стабильную версию, доступную по состоянию на апрель 2014 года.

Да, вы можете создать тему программно через API Kafka. И да, вы можете указать желаемое количество разделов, а также коэффициент репликации для темы.

Обратите внимание, что недавно выпущенный Kafka 0.8.1+ предоставляет немного отличающийся API, чем Kafka 0.8.0 (который был использован Биком в его связанном ответе). Я добавил пример кода чтобы создать тему в Kafka 0.8.1+, чтобы ответить на вопрос Как мы можем создать тема в Kafka из IDE с использованием API, о которой упоминал Бикс.

Ответ 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();`