【原创】kafka admin源代码分析

时间:2022-01-28 16:34:32

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