21 | AbstractFetcherThread：拉取消息分几步？

你好，我是胡夕。我来公布上节课的“课后讨论”题答案啦～ 上节课，咱们结合源码重点了解了Broker是如何处理延时请求的。课后我挑了advanceClock方法中的一个语句让你去分析它的目的。这行语句计算你的事是watcherList中总的已完成的请求数。另外在遍历watcherList的同时，代码还会将这些已完成的任务从链表中移除。 okay，你同意这个说法吗？或者说你有其他的看法吗？我们可以一起讨论下。

handlePartitionsWithErrors方法简单来说就是，如果可能出现leader切换，延迟处理带错误的partitions，并且延迟处理过程加了中断优先处理的锁，

打扰下大佬，最近集群出现isr抖动的问题，定位是leaderisrupdatelock的锁竞争导致的。除了正常的业务请求，我发现每次出现问题的罪魁祸首是follower节点的fetchfollower请求。 在测试环境发现fetchfollower的量受到集群内处理请求的节点数以及produce与consume流量的强影响。 我有以下两个疑惑： - 为什么consume流量上涨会导致fetchfollower的上涨？ - 我尝试扩容。集群内的topic一半AR设置为1 2 3，一半设置为4，5，6。我理解相当于均摊了partition数量但是没有为当前节点引入更多相关broker，按理说相当于拆分成两个集群，但是fetchfollower的数量仍然产生了近两倍的fetchfollower量。请问下具体原因是什么？ 与fetchfollower量相关的broker参数我找到了config限流，replica.fetch.wait.max.ms, fetch.min.bytes。但是这些参数都将加大丢消息以及被踢出isr的风险。如果您有别的方法来减少fetchfollower的量，将感激不尽。谢谢

老师，sourceBroker 这个配置是从 Zookeeper /broker/ids/[0-n] 这个节点下拿的配置吧，如果配置了 advertised.listener 那就会变成这个配置的地址，是这样吗？我现在遇到的问题就是内外网非双网卡的情况，多副本的 Topic 发送消息就会导致 Broker 报错。

handlePartitionsWithErrors 方法的实现原理 1、当partitions不为空时，调用delayPartitions方法 2、获取每个分区的partitionStates，如果当前分区状态不是isDelayed则将该分区移动至末尾，进行延迟处理

handlePartitionsWithErrors方法源码如下： // 处理可能由于领导层变更而导致的错误分区 private def handlePartitionsWithErrors(partitions: Iterable[TopicPartition], methodName: String): Unit = { if (partitions.nonEmpty) { debug(s"Handling errors in $methodName for partitions $partitions") delayPartitions(partitions, fetchBackOffMs) } } 底层调用delayPartitions方法 def delayPartitions(partitions: Iterable[TopicPartition], delay: Long): Unit = { partitionMapLock.lockInterruptibly() try { for (partition <- partitions) { Option(partitionStates.stateValue(partition)).foreach { currentFetchState => if (!currentFetchState.isDelayed) { partitionStates.updateAndMoveToEnd(partition, PartitionFetchState(currentFetchState.fetchOffset, currentFetchState.lag, currentFetchState.currentLeaderEpoch, Some(new DelayedItem(delay)), currentFetchState.state)) } } } partitionMapCond.signalAll() } finally partitionMapLock.unlock() }