AdminUtils.createTopic API 可以连接多个 zookeeper 节点吗？答案

【问题标题】：Can AdminUtils.createTopic API connect to multiple zookeeper nodes?AdminUtils.createTopic API 可以连接多个 zookeeper 节点吗？
【发布时间】：2016-12-13 02:22:24
【问题描述】：

当尝试使用下面提到的 createTopics 函数创建主题时，我在多节点 kafka 集群中遇到了“复制因子：1 大于可用代理：在多节点集群上为 0”。我有 3 个 kafka 代理，我正在尝试为每个主题创建 2 个分区并将复制因子保持为 1。不知道为什么会出现此错误。在单节点设置（1 个 zookeeper 和 1 个 kafka 代理）中同样可以正常工作。

任何帮助将不胜感激/

错误：

kafka.admin.AdminOperationException: replication factor: 1 larger than available brokers: 0
at kafka.admin.AdminUtils$.assignReplicasToBrokers(AdminUtils.scala:117)
at kafka.admin.AdminUtils$.createTopic(AdminUtils.scala:403)
at kafka.admin.AdminUtils.createTopic(AdminUtils.scala)
at io.confluent.examples.producer.ZookeeperUtil.createTopics(ZookeeperUtil.java:98)
at io.confluent.examples.producer.ProducerGroup.<init>(ProducerGroup.java:50)
at io.confluent.examples.producer.ProducerGroup.main(ProducerGroup.java:124)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:297)
at java.lang.Thread.run(Thread.java:745)






private static final int DEFAULT_SESSION_TIMEOUT = 10 * 1000;
private static final int DEFAULT_CONNECTION_TIMEOUT = 8 * 1000;
private static final String ZOOKEEPER_CONNECT = "zNode01:2181,zNode02:2181,zNode03:2181";

/**
 * Opens a new ZooKeeper client to access the Kafka broker.
 */
private static ZkClient connectToZookeeper ()
{
    return new ZkClient(ZOOKEEPER_CONNECT,
                        DEFAULT_SESSION_TIMEOUT,
                        DEFAULT_CONNECTION_TIMEOUT,
                        ZKStringSerializer$.MODULE$);
}

/**
 * Given a ZooKeeper client instance, accesses the broker and returns
 * information about Kafka's contents.
 *
 * @param zookeeperClient A ZooKeeper client to access broker information
 *                        through.
 */
private static ZkUtils zookeeperUtility (ZkClient zookeeperClient)
{
    boolean isSecureCluster = false;
    return new ZkUtils(zookeeperClient, 
                       new ZkConnection(ZOOKEEPER_CONNECT),
                       isSecureCluster);
}


public static void createTopics (ArrayList<String> names, int partitions, int replication)
{
    ZkClient zkClient = connectToZookeeper();
    ZkUtils zkUtils = zookeeperUtility(zkClient);

try{
    for (String name: names)
    {
        if (existsTopic(name))
            continue;

        AdminUtils.createTopic(zkUtils, name, partitions, replication, new Properties(),RackAwareMode.Disabled$.MODULE$);

    }

} catch (Exception ex) {
    ex.printStackTrace();
} finally {
    if (zkClient != null) {
        zkClient.close();
    }
}
}

我按照以下说明设置了多节点 kafka 集群。

设置多节点 Apache ZooKeeper 集群

在集群的每个节点上，将以下行添加到文件 kafka/config/zookeeper.properties

    server.1=zNode01:2888:3888
    server.2=zNode02:2888:3888
    server.3=zNode03:2888:3888
    #add here more servers if you want
    initLimit=5
    syncLimit=2

在集群的每个节点上，在 dataDir 属性表示的文件夹中创建一个名为 myid 的文件（默认情况下，该文件夹是 /tmp/zookeeper ）。 myid 文件应该只包含 znode 的 id（'1' 代表 zNode01，'2' 代表 ZNode02，等等......）

设置多代理 Apache Kafka 集群

在集群的每个节点上修改从文件 kafka/config/server.properties 中修改属性 zookeeper.connect：

    zookeeper.connect=zNode01:2181,zNode02:2181,zNode03:2181

在集群的每个节点上修改文件 kafka/config/server.properties 中的属性 host.name： host.name=zNode0x

在集群的每个节点上修改文件 kafka/config/server.properties 中的属性 broker.id（集群中的每个代理都应该有一个唯一的 id）

【问题讨论】：

你所有的卡夫卡经纪人都在设置中？
是的，他们都起来了！
正如@Knight71 所提到的，您的经纪人很可能没有启动。在运行代码之前，请尝试通过 kafka-topics.sh 创建主题并通过 kafka-console-producer.sh 生成主题。
我尝试从其中一个 zookeeper 节点的终端创建主题，它运行良好，所以我不确定发生了什么，如果我将 zookeeper.connect 与一个节点一起使用，它可以工作。只有当我在 Zookeeper.conenct 属性中有多个 zookeeper 节点时，它才会进入这个状态。
我的所有代理和我的 kafka 消费者都在所有 3 个节点中，我想如果 AdminUtils.createTopic api，可以连接到多个动物园管理员？

标签： java apache-kafka apache-zookeeper kafka-producer-api

【解决方案1】：

错误是说你的代理没有运行，而不是你不能为客户端使用多个 Zookeeper 地址（是的，你可以）

在较新版本的 Kafka 中，您可以使用 AdminClient 和 bootstrap.server，而不是 zookeeper.connect

【讨论】：