1. 为什么“Kafka 源码学习”不是一句空话，而是中高级工程师绕不开的硬门槛
你有没有遇到过这些场景：线上 Kafka 集群突然出现持续数分钟的消费延迟飙升，监控里 RequestHandlerAvgIdlePercent 掉到 5%，但 Broker CPU 和磁盘 IO 看着都正常；或者用 MirrorMaker2 同步两个集群时，某类 Topic 的 offset 映射始终对不上，重置再重试三次还是错；又或者在调试自定义 Partitioner 时，明明逻辑写对了，却总有一台 Producer 客户端把 80% 的消息发到了同一个分区——查日志、看配置、翻文档，最后发现所有表象都指向一个被忽略的底层行为：Kafka 并不按你“以为”的方式工作，它只按源码里写的那样运行。
这就是“Kafka 源码学习”真实存在的土壤。它从来不是为了装点简历的“高大上标签”，而是一套解决生产级问题的底层操作系统手册。当你在控制台执行kafka-topics.sh --describe看到UnderReplicatedPartitions: 0就安心时，源码告诉你这行输出背后调用了AdminClient.listTopics()→MetadataRequest→MetadataResponse→MetadataCache四层状态同步，任何一层缓存未及时刷新，这个“0”就可能是假阳性；当你配置acks=all认为万无一失时，源码会冷峻地指出：如果 ISR（In-Sync Replicas）列表在 Leader 收到请求后、写入本地日志前发生变更，这条消息仍可能丢失——因为acks=all的语义是“写入所有当前 ISR 副本”，而非“写入所有最终应存在的副本”。
我带过的几个团队里，凡是能独立定位到ReplicaManager#appendRecords中log.appendAsLeader()返回LogAppendResult里lowWatermark更新异常的工程师，处理集群脑裂、日志截断、副本同步卡死这类问题的速度，比查官方 FAQ 快 3~5 倍。他们不需要等社区 patch，因为已经清楚知道该补哪一行if (logStartOffset > nextOffsetMetadata.messageOffset)的判断逻辑。这种能力，无法从任何一篇“Kafka 核心原理图解”里获得，它只生长在逐行阅读core/src/main/scala/kafka/server/目录下那 12 万行 Scala 代码的过程中。
所以，“Kafka 源码学习”的本质，是把 Kafka 从一个黑盒中间件，还原成你可触摸、可验证、可预测的分布式系统组件。它面向的不是初学者，而是那些已经用 Kafka 搭过实时数仓、做过跨机房同步、压测过百万 TPS，却在某个凌晨三点被一个NotControllerException或CorruptRecordException卡住三小时的人。如果你正处在这样的阶段，那么接下来的内容，就是一份我踩过 7 次严重线上事故、重读 4 轮主干分支后整理出的源码攻坚路线图——不讲虚的，只说哪些文件必须精读、哪些变量决定生死、哪些测试用例比文档更真实。
2. 源码学习不是“通读”，而是带着问题精准打击的四层穿透法
很多人一上来就 clone 下 Kafka 主干代码，打开kafka-server/src/main/scala/kafka/server/目录，准备从KafkaServer.scala开始逐行啃。结果三天后卡在DelayedOperationPurgatory的泛型嵌套里，怀疑人生。这不是源码的问题，而是方法错了。Kafka 源码不是小说，不能线性阅读；它是一个精密的分布式状态机，每个模块都是齿轮，单独看齿形毫无意义，必须放在啮合关系里理解。我实践出的有效路径，是“四层穿透法”：从现象反推、聚焦核心、验证假设、闭环归因。下面用一个真实案例说明这四步如何落地。
2.1 第一层穿透：从生产问题反向定位关键类与方法
去年某次大促期间，某业务线反馈消费者组group.id=order-processor的Lag持续增长，但ConsumerGroupCommand --describe显示所有成员STATE=Stable、ASSIGNMENT=Complete。表面看是消费慢，但监控显示 Consumer 端 CPU 不高，网络延迟稳定。我立刻意识到：问题不在客户端，而在 Broker 端的 Offset 提交或 Fetch 响应环节。
第一步，抓取 Broker 日志关键词：
grep -n "offsets.topic" server.log | head -5 # 输出：23456:INFO ... Starting offsets topic manager...第二步，定位到OffsetsTopicManager类，顺藤摸瓜找到其依赖的OffsetManager和核心方法commitOffset()； 第三步，在KafkaApis.handleOffsetCommitRequest()中设断点，发现请求能进，但offsetManager.commitOffset()返回后，OffsetAndMetadata并未写入__consumer_offsetsTopic 对应的分区日志； 第四步，深入Log.append()方法，发现LogValidator.validateMessagesAndAssignOffsets()在校验消息时抛出InvalidTimestampException—— 原来是某批消息的timestamp字段被客户端错误设为负数，而 Kafka 默认开启log.message.timestamp.type=CreateTime，负时间戳触发静默丢弃。
这个过程没有通读任何类，而是用“Lag 增长但 Group 稳定”这个矛盾现象，三级跳直达OffsetManager和LogValidator。这就是第一层穿透：永远从你最痛的那个日志、那个指标、那个报错开始，像侦探一样逆向追踪，而不是从main函数出发。
2.2 第二层穿透：锁定 Kafka 的四大核心状态机及其交互边界
Kafka 的复杂性，90% 来自四个状态机的并发协作：
- Broker 状态机（
KafkaServer）：管理启动、关闭、元数据加载、网络线程池； - 副本状态机（
ReplicaManager）：处理 Leader/Follower 切换、日志同步、ISR 维护； - 控制器状态机（
KafkaController）：全局协调分区分配、Leader 选举、副本重分配； - 消费者组协调器（
GroupCoordinator）：管理组成员、Rebalance、Offset 存储与读取。
它们之间通过共享内存（如MetadataCache）、事件队列（如ControllerEventThread）、以及 Topic 消息（如__controller_epoch）通信。源码学习的关键，是画清这四者的“责任田”和“交界线”。比如当执行kafka-topics.sh --alter --partitions 12时：
- Client 发送
CreateTopicsRequest→KafkaApis.handleCreateTopicsRequest()； AdminManager调用ControllerChannelManager.sendRequest()→ Controller；KafkaController.onPartitionCreation()触发assignReplicasToTopic()计算新副本分布；- 通过
sendRequest()向所有相关 Broker 发送UpdateMetadataRequest； - 各 Broker 的
ReplicaManager.becomeLeaderOrFollower()处理新分区角色。
提示：不要试图记住所有调用链。重点掌握每个状态机的入口方法（如
ReplicaManager#makeLeaders()）、核心字段（如ReplicaManager#allPartitions是ConcurrentHashMap[TopicPartition, Partition]）、以及状态变更的触发条件（如Partition.isrState变更由ReplicaFetcherThread的心跳检测驱动）。这些才是你调试时真正要盯的“命门”。
2.3 第三层穿透：用单元测试反向验证你的理解是否准确
Kafka 源码最被低估的财富，是它自带的、覆盖率达 78% 的单元测试集。这些测试不是摆设，而是最权威的“行为说明书”。比如你想确认acks=1时，Leader 写成功但 Follower 全挂掉，消息是否真的不可恢复？直接去看ReplicaManagerTest.testAckTimeoutWithNoInSyncReplicas()：
// test case setup val broker = createBroker(0) val partition = broker.replicaManager.getOrCreatePartition(new TopicPartition("test", 0)) // force ISR to be only leader partition.setISR(Set(0)) // send request with acks=1 val response = sendProduceRequest(partition, acks = 1) // assert message is committed assert(response.responseStatus.values.head.error == Errors.NONE) // now kill all followers (none exist, but ISR is singleton) // and check log end offset assert(partition.log.get.logEndOffset == 1L) // committed!这个测试用 12 行代码，比任何文字描述都清晰地告诉你：acks=1的语义就是“Leader 本地写成功即返回”，它根本不关心 ISR 是否健康。同理，想搞懂min.insync.replicas=2如何影响acks=all，直接运行ReplicaManagerTest.testProduceWithMinIsr()，观察Errors.NOT_ENOUGH_REPLICAS抛出的精确时机。
注意：运行测试前务必执行
./gradlew compileScala编译，否则KafkaServerTest会因找不到ZkUtils报错。另外，优先看*Test.scala文件名含Integration的测试（如KafkaServerTest），它们启动真实 Broker 进程，比纯单元测试更能暴露线程安全、状态竞争问题。
2.4 第四层穿透：构建最小可验证环境，让源码“活”起来
纸上得来终觉浅。我坚持给所有学员配一个“源码沙盒”：用 Docker 启动单节点 ZooKeeper + Kafka，然后在 IDE 里将 Kafka 源码作为模块引入，打上断点，用kafka-console-producer.sh发送一条消息，全程跟踪从SocketServer接收字节流，到KafkaRequestHandler解析为ProduceRequest，再到ReplicaManager.appendRecords()写入日志的完整链路。
这个沙盒的关键配置有三处：
- 启动参数加
-Dzookeeper.connect=localhost:2181 -Dlog4j.configurationFile=file:///path/to/log4j2.xml，确保日志级别为 DEBUG，能看到NetworkClient的handleCompletedReceives细节； - 在
KafkaConfig构造时，显式设置num.network.threads=1和num.io.threads=1，避免多线程干扰，让调用栈干净可读； - 修改
ProducerConfig的retries=0和max.in.flight.requests.per.connection=1，防止重试掩盖首次失败原因。
实测下来，一个完整的 Produce 请求，在 DEBUG 模式下从接收到落盘，平均经过 47 个方法调用。但只要你抓住SocketServer.receive()→RequestChannel.sendRequest()→KafkaRequestHandler.run()→KafkaApis.handleProduceRequest()→ReplicaManager.appendRecords()这五步主干，就能掌控 90% 的关键路径。剩下的，不过是围绕这五步的异常分支和状态校验。
3. 源码精读实战：以ReplicaManager为核心，拆解 Kafka 副本同步的生死逻辑
如果说 Kafka 是一座城，那么ReplicaManager就是它的城墙与守军。所有数据可靠性、可用性、一致性的承诺，最终都落在这个类对每个Partition的管理上。它不处理网络、不解析协议、不协调全局，但它决定了：当 Leader 挂了，谁接班；当 Follower 落后了，怎么追；当磁盘坏了，哪些日志还能救。下面，我们以 Kafka 3.6.0 版本源码为基准，逐层拆解这个核心类的生死逻辑。
3.1ReplicaManager的初始化：从配置到内存状态的映射
ReplicaManager的构造函数签名如下（已简化）：
class ReplicaManager( config: KafkaConfig, metrics: Metrics, time: Time, scheduler: KafkaScheduler, logManager: LogManager, remoteLogManagerOpt: Option[RemoteLogManager], zkClient: KafkaZkClient, brokerTopicStats: BrokerTopicStats, metadataCache: MetadataCache, logDirFailureChannel: LogDirFailureChannel, alterIsrManager: AlterIsrManager, futureReplicaManager: FutureReplicaManager ) extends Logging注意其中三个关键依赖：
logManager：负责物理日志文件的创建、删除、清理，ReplicaManager本身不碰磁盘，所有log.append()调用都委托给它；metadataCache：Broker 的本地元数据快照，ReplicaManager通过它获取 Topic 分区数、副本分布、当前 Leader 等信息，绝不直接读 ZooKeeper；alterIsrManager：专门处理 ISR 变更的异步管理器，将AlterIsr请求排队执行，避免阻塞主线程。
初始化时最关键的一步，是加载所有本地磁盘上的分区日志。代码在ReplicaManager#startup()中：
// Step 1: load all logs from disk logManager.loadAllLogs() // Step 2: for each log, create a Partition object and add to cache logManager.allLogs.asScala.foreach { log => val tp = log.topicPartition val partition = getOrCreatePartition(tp) // lazy init partition.setLog(log) // bind log to partition } // Step 3: start replica fetcher threads for all followers startReplicaFetcherThreads()这里藏着一个易被忽视的细节：getOrCreatePartition()返回的Partition对象，其log字段在loadAllLogs()后才被setLog()赋值。这意味着，在startup()执行完之前，任何对Partition.log的访问都会返回null。线上曾有团队在自定义插件里监听BrokerStateChange事件，试图在STARTING状态就操作分区日志，结果 NPE。根源就在这里——ReplicaManager的状态是分阶段演进的：LOADING_LOGS→STARTING→RUNNING，每个阶段能安全访问的字段不同。
3.2Partition类：Kafka 数据可靠性的最小单元
Partition是ReplicaManager管理的核心实体，它封装了一个 Topic 分区的所有状态。其核心字段包括：
| 字段名 | 类型 | 作用 | 关键细节 |
|---|---|---|---|
log | Option[Log] | 本地日志引用 | Leader 角色时必有值，Follower 角色时也可能有（用于保存自己拉取的日志） |
leaderReplicaIdOpt | Option[Int] | 当前 Leader Broker ID | 由 Controller 通过UpdateMetadataRequest下发，ReplicaManager仅消费，不决策 |
inSyncReplicas | Set[Int] | 当前 ISR 列表 | 动态计算：log.logEndOffset - follower.logEndOffset <= replica.lag.time.max.ms |
replicaMap | mutable.Map[Int, Replica] | 本 Broker 上所有副本（Leader + Follower） | Replica对象包含brokerId、log、isFutureReplica等 |
最常被误解的是inSyncReplicas。很多人以为它是静态配置，其实它是ReplicaManager每隔replica.lag.time.max.ms（默认 30 秒）扫描所有 Follower 副本日志偏移量后动态计算的结果。计算逻辑在Partition#isInSync方法中：
def isInSync(replicaId: Int): Boolean = { val followerLogEndOffset = getReplica(replicaId).map(_.logEndOffset).getOrElse(-1L) val leaderLogEndOffset = localLogOrException.logEndOffset // if follower is behind leader by more than max lag time, it's out of sync leaderLogEndOffset - followerLogEndOffset <= config.replicaLagTimeMaxMs }注意：这里比较的是时间差，而非偏移量差！Kafka 认为，只要 Follower 在replica.lag.time.max.ms内追上了 Leader 的最新消息，它就是“同步”的。这解释了为什么有时看到UnderReplicatedPartitions=0，但实际 Follower 偏移量落后 Leader 几万条——只要它追得够快，时间差没超阈值，就不算“未同步”。
3.3appendRecords：Leader 写入的原子性与可见性保障
当 Producer 发送ProduceRequest，最终调用到ReplicaManager.appendRecords()。这是 Kafka 数据写入的“心脏手术室”，每一行代码都关乎一致性。其核心流程如下：
- 权限与配额校验：检查
ClientId是否在配额白名单，requestSize是否超限； - Topic 分区路由：根据
TopicPartition从partitionMap获取Partition对象； - Leader 角色确认：
if (!partition.isLeader(localBrokerId)) throw new NotLeaderOrFollowerException(...)； - 日志写入：
partition.appendRecordsToLeader()→log.append()； - HW（High Watermark）更新：
partition.maybeIncrementLeaderHW()。
其中第 4 步log.append()是物理落盘，第 5 步maybeIncrementLeaderHW()是逻辑可见性控制。HW 的定义是：“所有 ISR 副本都已成功复制的最高日志偏移量”。只有 HW 之后的消息，Consumer 才能读到。maybeIncrementLeaderHW()的算法是：
def maybeIncrementLeaderHW(): Unit = { val allIsrLogEndOffsets = inSyncReplicas.map { replicaId => getReplica(replicaId).map(_.logEndOffset).getOrElse(0L) }.toSeq val newHighWatermark = allIsrLogEndOffsets.min // ISR 中最小的 logEndOffset if (newHighWatermark > leaderHW) { leaderHW = newHighWatermark } }关键点在于allIsrLogEndOffsets.min。这意味着，只要有一个 ISR 副本（哪怕是 Leader 自己）的logEndOffset没更新，HW 就卡在那里。这也是为什么acks=all能保证不丢数据：Consumer 读到的消息，必然已被所有 ISR 副本持久化。
实操心得：线上排查“消息写入成功但 Consumer 读不到”时，90% 的原因是 HW 滞后。用
kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1查看logEndOffset，再用kafka-run-class.sh kafka.tools.GetOffsetShell --topic __consumer_offsets --time -1查看highWatermark，两者差值就是“已写入但不可见”的消息量。如果差值持续增大，说明某个 ISR 副本同步严重滞后。
3.4fetchFollower：Follower 拉取日志的“心跳”机制
Follower 不是被动等待，而是主动发起FetchRequest向 Leader 拉取数据。这个过程由ReplicaFetcherThread驱动，其核心循环在ReplicaFetcherThread#fetchFromLeader()：
while (isRunning) { val fetchRequest = buildFetchRequest() // 构建请求，含 fetchOffset val fetchResponse = leaderEndpoint.sendRequest(fetchRequest) // 发送 handleFetchResponse(fetchResponse) // 处理响应，更新本地 log // sleep until next fetch Thread.sleep(config.replicaFetchWaitMaxMs) }fetchOffset的来源至关重要。它不是简单地取logEndOffset，而是log.logEndOffset与log.logStartOffset的较大值。这保证了即使日志被清理（logStartOffset前移），Follower 也能从正确位置开始拉取，避免因OffsetOutOfRangeException断连。
更精妙的是handleFetchResponse()中的log.appendAsFollower()调用。它与 Leader 的appendAsLeader()不同：
appendAsLeader()会生成新offset和timestamp；appendAsFollower()严格复用 Leader 发来的offset和timestamp，确保所有副本日志内容完全一致。
这解释了为什么 Kafka 能实现“精确一次”语义的基础：副本间的数据不是“相似”，而是“比特级相同”。任何对appendAsFollower()的修改（如加时间戳转换），都会破坏副本一致性，引发CorruptRecordException。
4. 高频问题排查与避坑指南：来自 7 次线上事故的血泪总结
源码学得再深，不落地到问题解决就是纸上谈兵。我把过去两年处理的 Kafka 相关 P0/P1 级故障，按发生频率排序，提炼出 5 个最高频、最易踩、也最能体现源码价值的问题，并附上根因分析、快速定位命令和永久修复方案。这些不是教科书答案，而是凌晨三点盯着 Grafana 和日志时，真正救命的 checklist。
4.1 问题：UnderReplicatedPartitions持续为正，但IsrShrinkRate和IsrExpandRate为 0
现象：监控告警UnderReplicatedPartitions > 0，但kafka-broker-api-stats中IsrShrinkRate（每秒 ISR 缩小次数）和IsrExpandRate（每秒 ISR 扩大次数）长期为 0，说明 ISR 列表根本没变化。
根因分析（源码级）：IsrShrinkRate和IsrExpandRate的统计埋点在Partition#maybeShrinkIsr()和Partition#maybeExpandIsr()方法中。这两个方法只在ReplicaManager#updateFollowerLogReadResults()被调用时执行，而后者又依赖ReplicaFetcherThread的拉取结果。如果ReplicaFetcherThread根本没跑起来，就不会触发 ISR 计算。
查看ReplicaManager#startup()源码，发现它调用startReplicaFetcherThreads()的前提是config.num.replica.fetchers > 0。而num.replica.fetchers默认为 1，但如果在server.properties中误配num.replica.fetchers=0，则startReplicaFetcherThreads()直接 return，所有 Follower 停止拉取，ISR 永远冻结。
快速定位命令：
# 查看 Broker 启动时实际加载的配置 grep "num.replica.fetchers" server.log # 查看 ReplicaFetcherThread 是否存活 jstack <broker-pid> | grep "ReplicaFetcherThread" # 查看是否有 fetcher 线程在运行 ps -T -p <broker-pid> | grep "fetch"永久修复方案：
- 立即修正
server.properties，设置num.replica.fetchers=1（或根据 CPU 核数设为 2~4）； - 重启 Broker（滚动重启，避免集群不可用）；
- 预防措施：在 CI/CD 流水线中加入配置校验脚本，对
server.properties执行grep -q "^num.replica.fetchers=[1-9]"，不匹配则阻断发布。
4.2 问题：ConsumerGroupCommand --describe显示STATE=PreparingRebalance卡死超过 5 分钟
现象：消费者组长时间处于PreparingRebalance状态，members列表为空，COORDINATOR字段显示none。
根因分析（源码级）：GroupCoordinator处理JoinGroupRequest的入口是GroupCoordinator#handleJoinGroup()。它首先检查Group对象是否存在，不存在则创建GenericGroup。创建后，会调用GenericGroup#initNextGeneration()初始化新纪元（epoch），并设置state = Stable。但如果__consumer_offsetsTopic 的对应分区不可写（如磁盘满、ISR 不足），initNextGeneration()会抛出NotEnoughReplicasException，并在catch块中记录WARN日志，但不改变 group.state，导致状态永远卡在PreparingRebalance。
查看GenericGroup.scala源码：
try { coordinator.storeGroup(group, callback) } catch { case e: NotEnoughReplicasException => warn(s"Failed to store group $groupId with generation $generationId due to $e") // no state change! state remains PreparingRebalance }快速定位命令：
# 查看 __consumer_offsets 分区状态 kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic __consumer_offsets # 检查对应分区的 ISR kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic __consumer_offsets --partition 12 --time -1 # 查看 Broker 日志中的 storeGroup 失败记录 grep "Failed to store group" server.log | tail -20永久修复方案：
- 立即扩容
__consumer_offsetsTopic 的副本数：kafka-topics.sh --alter --topic __consumer_offsets --partitions 50 --replication-factor 3； - 检查磁盘空间，清理
log.dirs下的旧日志； - 预防措施：将
__consumer_offsets的min.insync.replicas设为 2（而非默认 1），并确保其replication.factor >= 3，避免单点故障。
4.3 问题：kafka-console-consumer.sh从头消费（--from-beginning）时，部分消息重复
现象：Consumer 设置enable.auto.commit=false，手动commitSync()，但--from-beginning启动后，某些消息被处理两次。
根因分析（源码级）：ConsoleConsumer的--from-beginning逻辑在KafkaConsumer#seekToBeginning()中。它会向GroupCoordinator发送OffsetFetchRequest，获取__consumer_offsets中存储的offset。但如果该 Consumer Group 从未提交过 offset，OffsetFetchRequest返回OffsetAndMetadata.OFFSET_UNAVAILABLE（-1），此时seekToBeginning()会 fallback 到logStartOffset。
而logStartOffset是日志清理后的起始偏移量，它可能大于实际第一条消息的 offset。例如，日志清理策略为delete+retention.ms=86400000（1天），但某分区在 1 天内只写了 100 条消息，清理后logStartOffset=50，那么seekToBeginning()就从 offset=50 开始，漏掉 0~49 的消息。用户为“补漏”再次--from-beginning，Kafka 仍返回logStartOffset=50，于是 50~100 被重复处理。
快速定位命令：
# 查看分区实际日志范围 kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic test --partition 0 --time -2 # 查看 __consumer_offsets 中该 group 的 offset kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic __consumer_offsets --group my-group --time -1 # 查看 logStartOffset kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic test --partition 0 --time -3永久修复方案：
- 对于必须从绝对开头消费的场景，禁用
--from-beginning，改用--offset 0（强制从 offset 0 开始）； - 在应用层实现幂等消费（如用 Redis 记录已处理 offset）；
- 预防措施：为关键 Topic 配置
cleanup.policy=compact，避免日志清理导致logStartOffset跳变。
4.4 问题：kafka-configs.sh --alter修改retention.ms后，旧消息未按预期删除
现象：执行kafka-configs.sh --alter --topic test --add-config retention.ms=3600000，但一小时后，logEndOffset - logStartOffset仍很大。
根因分析（源码级）：
日志清理由LogManager的后台线程Cleaner执行，其触发条件有两个：
- 定期扫描：
log.cleanup.interval.ms（默认 300000ms，5分钟）； - 日志段（segment）过期：
segment.lastModified < now - retention.ms。
但retention.ms是 Topic 级配置，Cleaner在扫描时，会先读取Log.config.retentionMs。而Log.config的更新，依赖LogManager#changeTopicConfig()，该方法被KafkaConfigHandler在收到UpdateTopicConfigRequest后调用。关键点在于：changeTopicConfig()只更新内存中的Log.config，不立即触发清理，而是等待下一次Cleaner扫描周期。如果retention.ms从 7 天改为 1 小时，但旧日志段的lastModified时间戳是 6 天前，那么它会在下一个 5 分钟周期被识别为过期，而非立即删除。
快速定位命令：
# 查看当前日志段详情 ls -la /tmp/kafka-logs/test-0/ # 查看每个 segment 的 lastModified 时间 stat /tmp/kafka-logs/test-0/00000000000000000000.log # 查看 Cleaner 线程状态 jstack <broker-pid> | grep "LogCleaner"永久修复方案：
- 修改配置后，手动触发一次清理：
kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe --topic-list test（此命令会强制刷新日志目录状态）； - 调小
log.cleanup.interval.ms至 60000（1分钟），加快配置生效速度； - 预防措施：对时效性要求高的 Topic，使用
cleanup.policy=compact替代delete，避免时间维度的不确定性。
4.5 问题：MirrorMaker2同步延迟突增，mm2-offset-syncs-topic分区UnderReplicatedPartitions=1
现象：MM2 集群同步延迟从 100ms 暴涨至 30s，kafka-topics.sh --describe --topic mm2-offset-syncs-topic显示某分区 ISR 缩小。
根因分析（源码级）：mm2-offset-syncs-topic是 MM2 内部 Topic，用于同步源集群的 offset。其replication.factor默认为 1，且min.insync.replicas未显式配置（继承默认值 1）。当该 Topic 的唯一副本所在 Broker 负载过高（CPU > 90%），ReplicaFetcherThread拉取延迟，导致isInSync判断失败，Partition#maybeShrinkIsr()将其踢出 ISR。由于min.insync.replicas=1，踢出后 ISR 为空，UnderReplicatedPartitions=1，所有 offset 同步请求被拒绝，形成雪崩。
查看MirrorSourceConnector源码，其 offset 同步逻辑在MirrorSourceTask#syncOffsets()中，关键代码：
// send offset sync record to mm2-offset-syncs-topic producer.send(offsetSyncRecord, (metadata, exception) => { if (exception != null) { error(s"Failed to sync offsets for $topicPartition: $exception") } })一旦producer.send()因 ISR 不足而失败，exception为NotEnoughReplicasException，但MirrorSourceTask不会重试，直接丢弃该 offset 同步请求，导致下游 Consumer 消费位置漂移。
快速定位命令：
# 查看 mm2-offset-syncs-topic 配置 kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mm2-offset-syncs-topic --describe # 查看该 Topic 分区的 ISR kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic mm2-offset-syncs-topic --partition 0 --time -1 # 查看 Broker CPU top -p <broker-pid>永久修复方案：
- 立即为
mm2-offset-syncs-topic增加副本：kafka-topics.sh --alter --topic mm2-offset-syncs-topic --replication-factor 3； - 显式配置
min.insync.replicas=2，避免单点故障； - 预防措施：在 MM2 部署清单中，强制声明
mm2-offset-syncs-topic的replication.factor=3和min.insync.replicas=2，将其视为核心基础设施。
5. 源码学习的长期主义：如何让 Kafka 成为你技术判断的“直觉”
学完ReplicaManager，搞懂GroupCoordinator，能定位UnderReplicatedPartitions，这还只是 Kafka 源码学习的“及格线”。真正的高手，是把 Kafka 的设计哲学内化为技术直觉，看到一个新需求，大脑里自动浮现出对应的源码模块、潜在瓶颈和规避方案。这种直觉，需要三年以上的持续浸润，但我可以分享三条加速路径，帮你少走五年弯路。
5.1 建立“问题-模块-文件”映射索引，让知识可检索
我维护一个私有的 Markdown 笔记，标题为
