kafka并发写大消息异常TimeoutException排查记录
作者:kl 发布时间:2023-11-27 23:07:11
前言
先简单介绍下我们的使用场景,线上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
来源:http://www.kailing.pub/article/index/arcid/259.html


猜你喜欢
- 如下所示:Ctrl+1或F2快速修复Ctrl+D快捷删除行Shift+Enter 快速切换到下一行,在本行的任何位置都可Ctrl+F11快速
- 一、背景spring-data-mongo 实现了基于 MongoDB 的 ORM-Mapping 能力,通过一些简单的注解、Query封装
- SimpleDateFormat是处理日期格式转换的类。官方API_1.8关于SimpleDateFormat继承于DateFormate截
- 承蒙各位厚爱,我们一起每天进步一点点!(鼠标选中空白处查看答案)1、现有如下代码段:x = 2;while(x<n/2){x = 2*
- 简介自从JDK中引入了stream之后,仿佛一切都变得很简单,根据stream提供的各种方法,如map,peek,flatmap等等,让我们
- 问题: 在win10系统中,使用IDEA 查看源码的时候,经常跳转代码,这样会让使用者感觉和方便,但是当你想要返回上次跳转的位置时,却不知道
- 实例如下:private void Form1_Load(object sender, EventArgs e)
- Token 一定要放在请求头中吗? 答案肯定是否定的,本文将从源码的角度来分享一下 spring security oauth2 的解析过程
- 目录环境依赖数据源方案一 使用 Spring Boot 默认配置方案二 手动创建脚本初始化使用 JdbcTemplate 操作实体对象DAO
- 属性问题引入前面我们已经在父工程中的dependencyManagement标签中对项目中所使用的jar包版本进行了统一的管理,但是如果在标
- 在Android TV上一般选中某个View, 都会有焦点突出放大的效果, 但是当在RecyclerView中(ListView或GridV
- Elastic Search是一个开源的,分布式,实时搜索和分析引擎。Spring Boot为Elasticsearch及Spring Da
- 产生背景为了解决异步线程产生的回调地狱//传统回调方式api.login(phone,psd).enquene(new Callback&l
- goto在Java中是一个保留字,但在语言中并没有用到它;Java没有goto。但是,Java也能完成一些类似于跳转的操作,主要是依靠:标签
- 前言我们之前实现了打包发布NuGet,但是发布后的引用是公有的,谁都可以访问,显然这种方式是不可取的。命令版本:10分钟学会Visual S
- InheritableThreadLocal的作用: 当我们需要在子线程中使用父线程中的值得时候我们就可以像使用ThreadLocal那样来
- 当目标数据库不能直连的,需要一个服务器作为中间跳板的时候,我们需要通过SSH通道连接数据库。ps:使用ssh连接,相当于本地开了个端口去连接
- 功能需求 最近项目中有这么一个功能,用户登录系
- 需要装一个插件:File - Settings- Plugins - 搜索gson 安装GsonFromat;如下两张图安装完成后 ,新建一
- 首先在pom文件里引入mqtt的依赖配置<!--mqtt--> <d