Kafka scala客户端在broker宕机对发送请求超时问题分析与方案

现象

生产中kafka集群一台服务器硬件故障下线，kafka集群具备高可用特性，下线broker上的leader分区自动切换到新的broker节点，客户端链接随之切换至新的节点继续提供服务，从流量上看也未发现异常情况；集群整体运行平稳，无异常。但后续有一个业务方反馈每10分钟就有少量接口响应时间升高导致超时，查看日志发现发送消息的接口有WARN日志，日志内容如下：

 

分析

         根据服务日志异常调用栈反映出的问题：业务线程在消息发送时，会触发获取远端metadata的服务请求，请求发到了宕机服务器后抛出Exception，并且很有周期性的每10分钟间隔发生少量异常；根据调用栈深入到kafka代码，找到发生问题的位置DefaultEventHandler.handle()：

红色方框里的代码，是发送消息前处理的一段逻辑，topicMetadataRefreshInterval为更新metadata路由表的间隔时间，缺省为600000(10分钟)。

 

发送消息前会判断本次发送是否需要更新路由表，判断依据为topicMetadataRefreshInterval指定的时间间隔, 设置不同值时执行情况如下：

topicMetadataRefreshInterval< 0 时：

         会跳过路由表更新逻辑，直接进入到消息发送环节（下图红色方框部分），如果发送不成功(比如partition leader切换情况)，则返回结果列表outstandingProduceRequests不为空，线程会睡眠retryBackoffMs指定的时间，然后同步发送brokerPartitionInfo.updateInfo()的请求,去获取失败topic的路由信息；同时本次发送失败，进入while循环判断是否内部重试，设置了重试次数那么内部继续重试发送，上一次更新了新的路由，重试后可能发送成功，否则继续之前的逻辑至到成功或重试次数结束返回；

 

topicMetadataRefreshInterval= 0 时：

         该情况下，每次发送消息前都会进入到路由表更新逻辑执行brokerPartitionInfo.updateInfo(…)

 

topicMetadataRefreshInterval> 0 时：

         该情况会按照topicMetadataRefreshInterval(缺省值为10分钟)间隔的时间，执行brokerPartitionInfo.updateInfo(…)，在10分钟之内的发送请求不会进入到该段逻辑；
 
上面文字描述了消息发送阶段，在什么条件下会执行brokerPartitionInfo.updateInfo(…)这段逻辑，而引起线程堵塞的真正凶手也将浮出水面，updateInfo()内部有段逻辑才是引起延迟的真凶fetchTopicMetadata()：

 
fetchTopicMetadata方法里，对broker列表首先做了混洗处理，然后从列表第一个broker进行遍历并执行同步发送metadata request请求，如果发向第一个broker获取meta失败，继续第二个直到成功或者列表遍历完毕退出；在生产案例中，一台broker宕机(整台机器断电)，如果请求的正好是该机器，建立TCP握手阶段会出现堵塞，直到socket的connection timeout超时返回，会造成业务线程堵塞；有个细节还需说明，如果机器存活进程死掉的情况下，该堵塞时间会小很多，原因TCP握手成功，访问服务进程失败会立即返回错误并切换下一个broker尝试；我们遇到的情况正好是IP不可访问，因而引起较长的超时等待，从而导致业务接口超时，但消息成功发送；

 
在发现kafka客户端有这样的行为后，为了防止影响，临时使用一台非标准配置的机器快速加入kafka集群解决问题（当时机房没有kafka标准配置的机器）。
 
 

方案

 针对以上情况，相关改进措施：

      1.   使用同步发送情况下，修改topic.metadata.refresh.interval.ms 值为负数；

             a. 客户端只有在发送失败的情况下(比如leader节点发生变化的分区)才会进行一次路由更新，不会对正常的发送链接产生影响；

             b. 发生路由更新会出现请求时间加长的情况，成功之后不会对后续发送产生影响；

             c. 但在partition数量扩容时，客户端正常发送情况下，缺乏感知分区变化进行动态调整；

      2.   在kafka最新java客户端中，采用了新的线程模型，即使在同步发送模式下也会采用异步线程来执行消息的发送，更新metadata的动作不会在业务线程中执行,而是异步sender线程来执行，并会跳过对失效节点的请求，避免网络问题引起线程的堵塞，同时接口层面采用了友好的超时中断机制，让业务更好的控制超时情况。