Kafka：从ZooKeeper获取代理主机

Question

出于特殊原因，我需要同时使用 - ConsumerGroup（又名高级消费者）和 SimpleConsumer（又名低级消费者）来读取 Kafka。对于 ConsumerGroup，我使用基于 ZooKeeper 的配置，并且对此完全满意，但 SimpleConsumer 需要实例化种子代理。

我不想同时保留动物园管理员和经纪人主机的列表。因此，我正在寻找一种方法，从ZooKeeper中自动发现特定主题的经纪人。

由于一些间接的信息，我相信这些数据存储在ZooKeeper中的以下路径之一：

< li > <代码>/经纪人/主题/

然而，当我试图从这些节点读取数据时，我得到了序列化错误(为此我使用了< code>com.101tec.zkclient):

组织。I0Itec.zkclient.exception.ZkMarshallingError： java.io.StreamCorruptedException： invalid stream header： 7B226A6D at org.I0Itec.zkclient.serialize.SerializableSerializer.deserialize（SerializableSerializer.java：37） at org.I0Itec.zkclient.ZkClient.derializable（ZkClient.java：740） at org.I0Itec.zkclient.ZkClient.readData（ZkClient.java：773） at org.I0Itec.zkclient.ZkClient.readData（ZkClient.java：761） at org.I0Itec.zkclient.ZkClient.readData（ZkClient.java：750） at org.I0Itec.zkclient.ZkClient.readData（ZkClient.java：744） ...64 省略 原因：java.io.StreamCorruptedException：无效的流标头：7B226A6D at java.io.ObjectInputStream.readStreamHeader（ObjectInputStream.java：804） at java.io.ObjectInputStream。（ObjectInputStream.java：299） at org.I0Itec.zkclient.serialize.TcclAwareObjectIputStream.（TcclAwareObjectIputStream.java：30） at org.I0Itec.zkclient.serialize.SerializableSerializer.deserialize（SerializableSerializer.java：31） ...还有 69 个

我可以毫无问题地编写和读取自定义Java对象（例如字符串），所以我相信这不是客户端的问题，而是编码的问题。因此，我想知道：

< li >如果这是正确的方法，如何正确读取这些节点？ < li >如果整个方法都是错误的，那么正确的方法是什么？

Answer 1

要使用shell执行此操作：

zookeeper-shell myzookeeper.example.com:2181
ls /brokers/ids
  => [2, 1, 0]
get /brokers/ids/2
get /brokers/ids/1
get /brokers/ids/0 

Answer 2

原来Kafka是用< code>ZKStringSerializer来读写数据到znodes的。因此，要修复这个错误，我只需将它作为最后一个参数添加到< code>ZkClient构造函数中:

val zkClient = new ZkClient(zkQuorum, Integer.MAX_VALUE, 10000, ZKStringSerializer)

使用它，我编写了几个有用的函数来发现经纪人ID、他们的地址和其他东西：

import kafka.utils.Json
import kafka.utils.ZKStringSerializer
import kafka.utils.ZkUtils
import org.I0Itec.zkclient.ZkClient
import org.apache.kafka.common.KafkaException


def listBrokers(): List[Int] = {
  zkClient.getChildren("/brokers/ids").toList.map(_.toInt)
}

def listTopics(): List[String] = {
  zkClient.getChildren("/brokers/topics").toList
}

def listPartitions(topic: String): List[Int] = {
  val path = "/brokers/topics/" + topic + "/partitions"
  if (zkClient.exists(path)) {
    zkClient.getChildren(path).toList.map(_.toInt)
  } else {
    throw new KafkaException(s"Topic ${topic} doesn't exist")
  }
}

def getBrokerAddress(brokerId: Int): (String, Int) = {
  val path = s"/brokers/ids/${brokerId}"
  if (zkClient.exists(path)) {
    val brokerInfo = readZkData(path)
    (brokerInfo.get("host").get.asInstanceOf[String], brokerInfo.get("port").get.asInstanceOf[Int])
  } else {
    throw new KafkaException("Broker with ID ${brokerId} doesn't exist")
  }
}

def getLeaderAddress(topic: String, partitionId: Int): (String, Int) = {
  val path = s"/brokers/topics/${topic}/partitions/${partitionId}/state"
  if (zkClient.exists(path)) {
    val leaderStr = zkClient.readData[String](path)
    val leaderId = Json.parseFull(leaderStr).get.asInstanceOf[Map[String, Any]].get("leader").get.asInstanceOf[Int]
    getBrokerAddress(leaderId)
  } else {
    throw new KafkaException(s"Topic (${topic}) or partition (${partitionId}) doesn't exist")
  }
}

Answer 3

这就是我的一位同事获取Kafka经纪人列表的方法。我认为当你想动态获取经纪人列表时，这是一个正确的方法。

下面是演示如何获取列表的示例代码。

public class KafkaBrokerInfoFetcher {

    public static void main(String[] args) throws Exception {
        ZooKeeper zk = new ZooKeeper("localhost:2181", 10000, null);
        List<String> ids = zk.getChildren("/brokers/ids", false);
        for (String id : ids) {
            String brokerInfo = new String(zk.getData("/brokers/ids/" + id, false, null));
            System.out.println(id + ": " + brokerInfo);
        }
    }
}

在由三个代理组成的集群上运行代码会导致

1: {"jmx_port":-1,"timestamp":"1428512949385","host":"192.168.0.11","version":1,"port":9093}
2: {"jmx_port":-1,"timestamp":"1428512955512","host":"192.168.0.11","version":1,"port":9094}
3: {"jmx_port":-1,"timestamp":"1428512961043","host":"192.168.0.11","version":1,"port":9095}