Kafka在Zookeeper中存储的信息有哪些？

0、总述

Kafka在Zookeeper中存储的信息都在/根路径下；大致分为5大类，

1. /brokers目录下存储着kafka集群broker的相关信息，包括：
   • /broker/ids/目录，采用临时znode的方式，存储所有的broker节点，每个broker的配置文件中都需要指定一个数字类型的id（全局不可重复）；
   • /broker/topics/目录，采用持久znode的方式，存储所有的topic注册信息；
   • /brokers/topics/[topic_name]/partitions/目录，采用持久znode的方式，存储某个topic的partitions所有分配信息；
2. /controller目录存储着Controller中央控制器所在kafka broker的信息；
   • 另外，在/controller_epoch路径下还存储着Controller的选举次数（默认从1开始），与partition状态信息中的值相同；
3. /consumers目录下存储着kafka集群所有Consumer的相关信息，包括：
   • /consumers/[group_id]/ids/目录，采用临时znode的方式，存储Consumer订阅信息，每个consumer都有一个唯一的ID用来标记消费者信息；
   • /consumers/[group_id]/owners/[topic_name]/[partition_id]目录下，采用临时znode的方式，存储着consumer group对应的各个topic及partition的消费者线程。
   • /consumers/[group_id]/offsets/[topic_name]/[partition_id]目录下，采用持久znode的方式，存储着consumer group对应的各个topic及partition的消费偏移量。
4. /admin目录下存储着kafka集群的管理信息，包括：
   • /admin/reassign_partitions目录下存储着partition重分配信息；
   • /admin/preferred_replica_election目录下存储着最优replica选举信息；
   • /admin/delete_topics目录下存储着近期删除的topic；
5. /isr_change_notification/目录下存储着ISR变更通知信息；

1） /brokers（Broker相关信息）

1> broker注册信息

路径：/broker/ids/[broker_id]

存储所有的broker节点，每个broker的配置文件中都需要指定一个数字类型的id（全局不可重复），此节点为临时znode。

存储数据示例：

[zk: localhost:2181(CONNECTED) 25] get /brokers/ids/0

{
	"listener_security_protocol_map": {
		"PLAINTEXT": "PLAINTEXT"
	},
	"endpoints": ["PLAINTEXT://192.168.1.76:9092"],
	"jmx_port": -1,
	"features": {},
	"host": "192.168.1.76",
	"timestamp": "1649662012051",
	"port": 9092,
	"version": 5
}

存储数据解析：

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

当Kafka集群中有新节点上线 或 旧节点下线时，这个路径(/broker/ids)下的数据就会改变；

2> topic注册信息

路径：/broker/topics/[topic_name]；

存储所有的topic注册信息；

存储数据示例：

[zk: localhost:2181(CONNECTED) 27] get /brokers/topics/saint-test-topic 

{
	"version": 2,
	"partitions": {
		"0": [0]
	},
	"adding_replicas": {},
	"removing_replicas": {}
}

存储数据解析：

• partitions：topic中各个partition的ID，以及其对应的ISR中各个broker的ID的列表。

当有topic被创建 / 删除，或者partition发生变更时，这个路径下的数据就会更新。通过对topic以及上述/broker/ids节点变更注册监听，可以实现producer的负载均衡。

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

3> partition状态信息

路径：/brokers/topics/[topic_name]/partitions/[partition_id]/state

存储某个topic的partitions所有分配信息；

存储数据示例：

[zk: localhost:2181(CONNECTED) 29] get /brokers/topics/saint-test-topic/partitions/0/state

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

存储数据解析：

• controller_epoch：controller的纪元（代数），即集群重新选举controller的次数。

• leader：该partition选举leader的brokerId。

• leader_epoch：partition leader的纪元（代数），即当前partition选举leader的次数。

• isr：同步副本组brokerId列表。

2） /controller（Controller注册信息）

路径：/controller

存储Controller中央控制器所在kafka broker的信息；

存储数据示例：

[zk: localhost:2181(CONNECTED) 31] get /controller

{
	"version": 1,
	"brokerid": 0,
	"timestamp": "1649662012132"
}

存储数据解析：

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

如果Controller信息节点被删除，就会触发集群重新选举Controller。

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

3） /consumers（Consumer订阅信息）

路径：/consumers/[group_id]/ids/[consumer_id]

存储Consumer订阅信息，每个consumer都有一个唯一的ID用来标记消费者信息，此节点是一个临时的znode；

存储数据示例：

[zk: localhost:2181(CONNECTED) 35] get /consumers/consumer-group-01/ids/1-d110015a-0317-466c-a510-c5d245d05209

{
  "version": 1,
  "subscription": {
    "saint-test-topic": 1
  },
  "pattern": "white_list",
  "timestamp": "1558617131642"
}

存储数据解析：

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

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

另外，在/consumers/[group_id]/offsets/[topic_name]/[partition_id]目录下，采用持久znode的方式，存储着consumer group对应的各个topic及partition的消费偏移量。
在/consumers/[group_id]/owners/[topic_name]/[partition_id]目录下，采用临时znode的方式，存储着consumer group对应的各个topic及partition的消费者线程。

4） /admin（集群管理信息）

1> partition重分配信息

路径：/admin/reassign_partitions

存储数据示例：

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

存储数据解析：

• topic：订阅的topic名称。
• partition：重新分配partition的id。
• replicas：partition副本所在的broker id。

2> 最优replica选举信息

路径：/admin/preferred_replica_election

存储数据示例：

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

存储数据解析：

• topic：订阅的topic名称。
• partition：重新分配partition的id。
• replicas：partition副本所在的broker id。

5）/isr_change_notification（ISR变更通知信息）

路径：/isr_change_notification/[isr_change_x]

各个partition的ISR集合并不是一成不变的。当ISR发生变化（比如有replica超时）时，controller会将发生变化的那个partition存入/isr_change_notification/[isr_change_x]中