kafka并发异常TimeoutException排查

时间:2022-05-23 kl 人气:6

前言

先简单介绍下我们的使用场景，线上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底层原理》

加载全部内容