controller如何选主及其启动过程分析broker启动 kafka.controller.KafkaController.startup() // controller启动 注册了启动Startup事件用于选择controller，即controller选主(KafkaController.scala 1141行) 以及RegisterBrokerAndReelec kafka.se

HW更新leader节点更新过程kafka.cluster.Partition.maybeIncrementLeaderHW(leaderReplica: Replica, curTime: Long) // Partition.scala 462行 这个方法的注释作者清楚的解释了HW增加的相关逻辑与背景 /** * Check and maybe increment the high

LEO更新leader节点更新过程写消息时直接更新LEO的代码// kafka/log/Log.scala /*641*/ private def append(records: MemoryRecords, isFromClient: Boolean, assignOffsets: Boolean, leaderEpoch: Int): LogAppendInfo = { // .

副本复制协议kafka的副本复制协议与目前公开的算法最相似的是PacificA。注意是相似。青老师跟我讲因为PacificA本身是push的模式，而kafka的消息复制是拉的模式。 There are a rich variety of algorithms in this family including ZooKeeper’s Zab, Raft, and Viewstamped Repli

数据写盘broker端消息正常写盘写盘Log.appendLogSegment.appendFileRecords.appendMemoryRecords.java.writeFullyTo 通过FileRecords的channel(FileChannel)字段完成写入 打开FileChannel打开过程 LogSegment.openFileRecords.open public

强制写盘与JDK对其的支持、mmap、零拷贝等系统层面系统调用fsync、fdatasync、sendfile、mmap fsync、fdatasync fsync, fdatasync - synchronize a file’s in-core state with storage device fsync() transfers (“flushes”) all modified in-

offset range查询的实现我们在实际使用过程中经常需要查询某个topic的某分区的offset的range命令行： kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list xxxx:9092 -topic xxxtopic --time -2 kafka-run-class.sh kafka.tools.Get

消费者组机制实现消费分组如何完成通过分组相关的rpc请求完成。以一个3分区的topic为例子，指定消费组消费后梳理请求日志如下： 先加入第一个消费者后节点0的请求处理日志： apiKey=FIND_COORDINATOR // coordinator_key=ss-t01-group,coordinator_type=0 apiKey=JOIN_GROUP // member_id=, ap

生产者批量发送机制实现 图加载有点慢，看不到的话可以点击查看大图

代码版本、编译与常用命令编译 下载sbt 下载scalaIDE 下载gradle 选择4版本，可以跳过一些问题 设置对应环境变量 gradle -b build.gradle jar 编译kafka代码 gradle -b build.gradle eclispe 生成eclipse的工程 然后导入到eclispe中, 导入后个别配置问题导致的报错fix一下就能没有错