kafka并发写大消息TimeoutException排查记录

原创
2019/09/20 14:25
阅读数 1K

前言

先简单介绍下我们的使用场景,线上5台Broker节点的kafka承接了所有binlog订阅的数据,用于Flink组件接收数据做数据中台的原始数据。昨儿开发反馈,线上的binlog大量报错,都是kafka的异常,而且都是同一条topic抛的错,特征也很明显,发送的消息体非常大,主观判断肯定是写入大消息导致的超时了,异常详情如下:

thread:  kafka-producer-network-thread | producer-1
throwable:  org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for BIN-LOG-DATA-center-dmz2-TABLE-kk-data-center-ods_contract_finance_info-0 due to 56352 ms has passed since last append

定位异常点

应用抛一个不常见的异常,一般操作是先去百度or谷歌搜索一番的,就上面这个timeout超时的异常,搜索引擎的结果都是producer连不上Borker导致的问题,根本不是我们这个场景的,所以其次我们就需要从源码中寻找答案了。博主使用的开发工具是IDEA,借助IDEA很容易定位到异常抛出点。首先定位TimeoutException异常类,然后按住ctrl键,点击这个类,会出现如下图所有抛TimeoutException异常的点,然后根据异常message内容,寻找相匹配的点击进去就是抛异常的地方了,如图,红色箭头所指即代码位置:

分析抛异常的逻辑

程序中的异常,一定是符合某些条件才会抛出的,想要解决异常,只要让运行时的环境不满足抛异常的条件即可,下面就是抛异常的代码:

    boolean maybeExpire(int requestTimeoutMs, long retryBackoffMs, long now, long lingerMs, boolean isFull) {
        if (!this.inRetry() && isFull && requestTimeoutMs < (now - this.lastAppendTime))
            expiryErrorMessage = (now - this.lastAppendTime) + " ms has passed since last append";
        else if (!this.inRetry() && requestTimeoutMs < (createdTimeMs(now) - lingerMs))
            expiryErrorMessage = (createdTimeMs(now) - lingerMs) + " ms has passed since batch creation plus linger time";
        else if (this.inRetry() && requestTimeoutMs < (waitedTimeMs(now) - retryBackoffMs))
            expiryErrorMessage = (waitedTimeMs(now) - retryBackoffMs) + " ms has passed since last attempt plus backoff time";

        boolean expired = expiryErrorMessage != null;
        if (expired)
            abortRecordAppends();
        return expired;
    }

可以看到,我们的异常是在第一个逻辑判断时候就满足了所以抛异常了。在此处有可能会抛出三个不同的timeout异常,用中文语义翻译条件分别是:

  • 没设置重试,并且发送批次(batch.size)满了,并且配置请求超时时间(request.timeout.ms)小于【当前时间减去最后追加批次的时间】
  • 没设置重试,并且配置请求超时时间(request.timeout.ms)小于【创建批次时间减去配置的等待发送的时间(linger.ms)】
  • 设置重试,并且配置请求超时时间(request.timeout.ms)小于【当前时间-最后重试时间-重试需要等待的时间(retry.backoff.ms)】

上面括号中的参数就是kafka producer中配置的相关的参数,这些参数都没有重新设置过,batch.size默认是10kb大小,而引发报错的消息都是36kb的大小,默认的request.timeout.ms超时设置是30s,所以在这个判断可能过期了的方法中,引发我们异常的主要原因是batch.size和request.timeout.ms的参数设置问题了。

真实原因-解决方案

从上面代码看表面原因是参数设置不够了,实际上呢,博主使用kafka-test启动了五个Borker集群做复现验证测试,测试写入相同的36kb的message,在所有配置也保持默认的情况下,也根本毫无压力。后面查找相关的错误日志,发现所有的TimeoutException集中在几乎同一时刻,经查明,是因为业务批量导入了数据到mysql中,造成binlog消息突然增加,高并发的往kafka写大消息导致Borker处理不过来,造成的TimeoutException超时,所以真正解决问题就可以从两个方面入手:

  • 服务端:增加Borker,并设置多个TopicPartition,平摊写入压力,这个是根本的解决问题
  • 客户端:加大request.timeout.ms、batch.size参数,或者开启消息重试,这种方案治标不治本,但是也能大概率的减少因为此类场景导致的TimeoutException

结语

异常不可怕,所有异常都是人为抛的,都是有既定的触发条件的,只要定位到触发异常的条件对症下药即可解决问题。不过博主五年来的经验发现,日志打印真的是门艺术,在这个方面,Spring框架和Dubbo以及Apollo配置中心框架就是日志打印的典范,不管发生什么异常,日志里都会输出详细的上下文环境,异常的原因,建议的解决方法,如果涉及到相关的配置,也会打印该怎么配置最好。反观kafka client的这条TimeoutException就显的信息量有点过少了,如果能把相关的配置信息和排查的方向写明会更好。最后安利一波kafka test,轻松搭建多Borker的kafka集群,一个注解就ok了。详情参考我的这篇博文《spring boot集成kafka之spring-kafka深入探秘》

作者简介:

陈凯玲,2016年5月加入凯京科技。现任凯京科技研发中心架构组经理,救火队队长。独立博客KL博客(http://www.kailing.pub)博主。

展开阅读全文
打赏
1
1 收藏
分享
加载中
当时, 生产的速度大概是多少 ?
06/05 16:36
回复
举报
你好,
我也遇到了同样的问题,
请问下,
在你的案例中, 出问题的Kafka集群是什么规模的 ? broker和patition的数量是多少 ?
06/05 16:19
回复
举报
更多评论
打赏
2 评论
1 收藏
1
分享
返回顶部
顶部