谈谈ZooKeeper中Kafka相关信息的存储体系

ZooKeeper（下文简称ZK）是应用广泛的分布式服务协调组件，它对于大数据领域的其他组件，如HDFS、YARN、HBase、Kafka等等，都扮演着基础角色。本文就来描述ZK如何存储与Kafka相关的信息。

以下是Kafka基本概念的名词，本文有可能涉及到，不再具体解释：

Broker（代理节点）、Controller（控制器）、Topic（主题）、Partition（分区）、Leader（分区领导者）、Follower（分区跟随者）、Replica（分区副本）、ISR（分区的同步副本组，即in-sync replicas）、Producer（生产者）、Consumer（消费者）、Consumer Group（消费组）、Offset（消费偏移量）

在kafka.utils.ZkUtils对象的开头，预先定义了很多ZK路径，列举如下。

  val AdminPath = "/admin"
  val BrokersPath = "/brokers"
  val ClusterPath = "/cluster"
  val ConfigPath = "/config"
  val ControllerPath = "/controller"
  val ControllerEpochPath = "/controller_epoch"
  val IsrChangeNotificationPath = "/isr_change_notification"
  val LogDirEventNotificationPath = "/log_dir_event_notification"
  val KafkaAclPath = "/kafka-acl"
  val KafkaAclChangesPath = "/kafka-acl-changes"

  val ConsumersPath = "/consumers"
  val ClusterIdPath = s"$ClusterPath/id"
  val BrokerIdsPath = s"$BrokersPath/ids"
  val BrokerTopicsPath = s"$BrokersPath/topics"
  val ReassignPartitionsPath = s"$AdminPath/reassign_partitions"
  val DeleteTopicsPath = s"$AdminPath/delete_topics"
  val PreferredReplicaLeaderElectionPath = s"$AdminPath/preferred_replica_election"
  val BrokerSequenceIdPath = s"$BrokersPath/seqid"
  val ConfigChangesPath = s"$ConfigPath/changes"
  val ConfigUsersPath = s"$ConfigPath/users"
  val ProducerIdBlockPath = "/latest_producer_id_block"

我们可以通过ZK命令行或可视化工具来观察这些路径下面的存储情况。为了方便梳理，将一些主要的节点画成树形结构，用下图表示。注意叶子节点颜色为蓝色表示临时（ephemeral）znode，黄色则表示永久（persistent）znode。

下面我们择重点解释这些ZK路径下的信息。

代理节点注册信息

路径为/brokers/ids/[broker_id]，其中存储的数据示例如下。

{
  "listener_security_protocol_map": {
    "PLAINTEXT": "PLAINTEXT"
  },
  "endpoints": ["PLAINTEXT://hadoop7:9092"],
  "jmx_port": 9393,
  "host": "hadoop7",
  "timestamp": "1554349917296",
  "port": 9092,
  "version": 4
}

• jmx_port：JMX端口号。
• host：所在主机名或IP地址。
• timestamp：启动时的时间戳。
• port：开放的TCP端口号。
• version：版本号。以下所有version值均是代表版本号，不再赘述。

当Kafka集群中有节点上下线时，这个路径下的数据就会更新。

主题注册信息

路径为/brokers/topics/[topic_name]， 其中存储的数据示例如下。

{
  "version": 1,
  "partitions": {
    "8": [103],
    "4": [109],
    "9": [104],
    "5": [110],
    "6": [111],
    "1": [106],
    "0": [105],
    "2": [107],
    "7": [102],
    "3": [108]
  }
}

• partitions：主题中各个分区ID及其对应的ISR中各个代理节点ID的列表。

当有主题被创建或删除，以及主题的分区发生变更时，这个路径下的数据就会更新。通过对主题以及上节所述节点变更注册监听，就能实现生产者的负载均衡。

另外，在/admin/delete_topics下还保存有已经标记为删除的主题名称（只有名称，没有其他数据）。在/config/topics/[topic_name]下保存有各个主题的自定义配置。

分区状态信息

路径为/brokers/topics/[topic_name]/partitions/[partition_id]/state，其中存储的数据示例如下。

{
  "controller_epoch": 17,
  "leader": 105,
  "version": 1,
  "leader_epoch": 2,
  "isr": [105]
}

• controller_epoch：控制器的纪元（代数），即集群重新选举控制器的次数。
• leader：当前分区领导的Broker ID。
• leader_epoch：分区领导的纪元（代数），即当前分区重新选举领导的次数。
• isr：该分区对应的ISR中各个代理节点ID的列表。

控制器注册信息

当前控制器信息的路径就是/controller，其中存储的数据示例如下。

{
  "version": 1,
  "brokerid": 104,
  "timestamp": "1554349916898"
}

• brokerid：现在集群中控制器的节点ID。
• timestamp：最近一次控制器变化的时间戳。

如果控制器信息节点被删除的话，就会触发集群重新选举控制器。ZK对选主可以说是有天然的支持。

另外，在/controller_epoch路径下还保存有控制器的纪元值，与分区状态信息中的值相同。每重新选举一次，该值就会加1。

消费者订阅信息

消费者本身的信息路径为/consumers/[group_id]/ids/[consumer_id]，其中存储的数据示例如下。

{
  "version": 1,
  "subscription": {
    "bl_mall_orders": 1
  },
  "pattern": "white_list",
  "timestamp": "1558617131642"
}

• subscription：订阅的主题名称及该主题对应消息流个数的映射。
• pattern：订阅方式，可取值为静态（static）、白名单（white_list）、黑名单（black_list）。
• timestamp：消费者创建时的时间戳。

通过ZK维护的消费组及消费者信息，可以实现消费者负载均衡。

另外，在/consumers/[group_id]/offsets/[topic_name]/[partition_id]下存储有消费组对应各个主题及分区的消费偏移量，在/consumers/[group_id]/owners/[topic_name]/[partition_id]下存储有消费组对应各个主题及分区的消费者线程。

最优副本选举信息

当由于节点宕机等原因使得分区领导变得不再均匀分布时，我们可以使用Kafka提供的kafka-preferred-replica-election工具重新将分区创建时的最优副本（前提是在ISR内）选举为分区领导者，也可以开启分区领导自动平衡的功能（auto.leader.rebalance.enable）。当正在选举最优副本时，ZK中就会创建/admin/preferred_replica_election节点，其中存储着需要调整最优副本的分区信息，示例数据如下。

{
  "version": 1,
  "partitions": [
    {
      "topic": "bl_mall_orders",
      "partition": 1
    },
    {
      "topic": "bl_mall_products",
      "partition": 0
    }
  ]
}

分区重分配信息

与上面的kafka-preferred-replica-election工具类似，Kafka还提供了kafka-reassign-partitions工具，但它的功能更为强大。它可以重新分配分区的所有分区（领导者和跟随者）位置，甚至更改副本因子。当集群扩容或跟随者分布也不均匀时，就可以利用它。该工具会生成JSON格式的重分配计划，并存入ZK中/admin/reassign_partitions节点，示例数据如下。

{
  "version": 1,
  "partitions": [
    {
      "topic": "bl_mall_wish",
      "partition": 1,
      "replicas": [0, 1, 3]
    }
  ]
}

ISR变更通知信息

分区的ISR集合并不是一成不变的。当ISR发生变化（如有副本超时）时，控制器会将发生变化的分区存入/isr_change_notification/[isr_change_x]中。目前暂时不容易找到它的数据格式，因此留空。