回顾一下kafka相关的概念:
Kafka Broker新建Topic的大致流程
- Kafka Topic Client发出创建Topic请求,到Zookeeper两个配置路径:/config/topics/[topic]和/brokers/topics/[topic]
- KafkaController 在 ZooKeeper 的 /brokers/topics 节点上注册 watcher,当 topic 被创建,则 TopicChangeListener执行监听。
- KafkaController读取/brokers/topics读取,创建Topic主要是分区的创建,只要分区创建成功了,那么Topic也创建成功了。
- Topic分区创建之后,会涉及到分区状态和副本状态的转化,set AR(副本)和设置一个副本成为ISR中的Leader
- 回写上面的分区状态和ISR中Leader状态到/brokers/topics/[topic]/partitions/[partition]/state
- 通过RPC 向相关的 broker 发送 LeaderAndISRRequest,主要是同步Partition状态,Leader,ISR信息。
Kafka的Broker删除Topic的大致流程
- Kafka Topic Client发出删除Topic请求,发送到Zookeeper中/admin/delted_topics
- KafkaController中函数DeleteTopicsListener监听/admin/delted_topics数据变化,会出发handleChildChange回调函数,会触发Topic的删除。Topic会进入TopicDeletionManager的待删除列表,真正调用删除是Topic删除的具体任务。删除Topic是一个异步任务的过程。删除Topic最终其实是删除Replica,删除完Replicat之后会清理Topic相关的Zookeeper的数据。执行删除Topic最开始注销监听,然后执行删除Replica等一系列的事情。
- 最后清理topic相关zookeeper的数据。这样topic就最终被删除。
Kafka的Producer写入过程
- Producer 先从 Zookeeper 带有 "/brokers/.../state"标识的节点找到该 partition 的Broker节点(Leader节点)
- Producer将消息发送给该leader节点
- Leader将消息写入本地Log
- Leader发送消息给Follower
- Followers 从Leader pull消息,写入本地 log 后给Leader发送ACK
- Leader收到所有ISR中的Replica的ACK 后,增加HW(high watermark)最后commit 的 offset)
- Leader向Producer发送ACK