【原创】kafka admin源代码分析

Love The Way You Lie 2022-03-31 18:50 186阅读 0赞

admin包定义了命令行的一些实现

**一、AdminOperationException.scala**

一个异常类，表示执行admin命令时候抛出的异常

**二、AdminUtils.scala**

admin一些常用工具方法：

1. assignReplicasToBrokers：负责分配副本到不同的broker上。主要有两个目标：① 尽可能均匀地在不同的broker上分配副本；② 对于被分配到同一个broker上的分区而言，它们的其他副本会尽可能地分配到不同的broker上。如果要达到这些目标，需要①分配每个分区的I一个副本时从broker列表中的一个随机位置开始按照round-bin方式分配；②使用一个递增位移来分配剩下的replica分配。需要注意的是，replica的factor不能大于broker个数

2. addPartitions：用于为一个已有的topic增加分区

3. deleteTopic：删除topic，就是在zookeeper上创建一个/admin/delete\_topics节点下创建一个子节点

4. topicExists：判断一个topic是否存在

5. createTopic：先分配副本，然后在zookeeper上对应的路径上创建topic

6. changeTopicConfig：更新一个已有topic配置信息并创建一个变更消息通知，这样这个变更可以传给其他broker

7. fetchTopicConfig：从zookeeper中读取topic的配置信息，具体路径为/config/topics/\[topic\]

**三、PreferredReplicaLeaderElectionCommand.scala**

该类将每个分区的leader移动到preferred replica上，主要用于在各个broker上重新再平衡leadership之用。既可以作为命令直接使用，也可以在指定属性auto.leader.rebalance.enable=true时由Kafka自动调用。默认是300秒自动检查一次。PreferredReplicaLeaderElectionCommand object的parsePreferredReplicaElectionData和writePreferredReplicaElectionData方法分别解析和构造zookeeper路径下/admin/preferred\_replica\_election节点的json数据串。

PreferredReplicaLeaderElectionCommaind class定义了两个方法：

1. validatePartition：验证分区合法性，主要检查分区是否存在。

2. moveLeaderToPreferredReplica：将领导权转到preferred副本上。具体逻辑如下：

*  过滤掉那些不存在的分区
 *  构造特定的JSON串更新到zookeeper的/admin/preferred\_replica\_election节点中
 *  如果启用了自动balance，会有一个后台线程定期地监听该节点数据的变化

**四、ReassignPartitionsCommand.scala**

向Kafka集群中添加新的服务器很容易，指定好唯一的broker id并启动起来就好了，但这些新加的服务器并不会被自动地分配新的数据分区，因此需要手动地将一部分数据搬到新加的服务器上。该类就是为实现这一目的。核心方法是reassignPartitions，逻辑就是在zookeeper的/admin\_reassign\_partitions下创建对应的数据记录，controller会注册一个zookeeper监听器监听该节点的数据变化。

**五、TopicCommand.scala**

一个可运行object，用于创建、删除、描述或变更topic：

1. createTopic：创建topic，值得注意的是，通过配置replica-assignment可以手动地分配哪些分区落在哪些broker上。如果不指定的话，Kafka自动帮你做分配的事情，然后创建topic

2. alterTopic： 修改topic

3. listTopic：列出当前集群中所有topic

4. deleteTopic：删除一个topic。其实就是标记该topic已经可以被删除

5. describeTopic：主要是获取该topic的分区信息、副本信息、leader 副本以及ISR