背景
某个应用需要配置 Kafka 集群信息,且需要在验证集群是否可达。基本实现思路是创建一个生产者对象,然后发送一条测试数据,调用 Producer
的 send
方法发送消息后,再调用 get()
方法,即同步发送消息,测试通过后再关闭 Producer
对象。
近期碰到了一个机器服务正常,异步消息能正常发送,但是同步一直报异常:
org.apache.kafka.common.errors.TimeoutException: Expiring 1 record(s) for topicxxx-0: 20024 ms has passed since batch creation plus linger time
问题排查
Kafka 状态正常,在机器 A 上使用生产者命令异步发送数据,Kafka服务器上消费数据能正常消费:
而应用通过同步发送 API 时一直报超时异常。
解决办法
调大生产者测试代码的超时时间:
//默认是30000ms
producerConfig.put("request.timeout.ms", "300000");
producerConfig.put("transaction.timeout.ms", "300000");
producerConfig.put("max.block.ms", "300000");
producerConfig.put("batch.size", "1048576");
producerConfig.put("linger.ms", "1");
producerConfig.put("buffer.memory", "33554432");
Kafka 生产者和消费者测试命令
对于包含认证的 Kafka 访问自带的生产者和消费者应用时,可以创建一个 config 配置文件 config.properties ,内容主要是 Kafka 地址和访问协议:
security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-256
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="xxx";
添加配置后,常用的三个命令也添加认证:
./kafka-topics.sh --bootstrap-server IP:PORT --list --command-config ../config/config.properties./kafka-console-producer.sh --bootstrap-server IP:port --topic test-topic --producer.config ../config/config.properties./kafka-console-consumer.sh --bootstrap-server IP:port --topic test-topic --consumer.config ../config/config.properties
注意点就是这三个脚本添加配置文件的参数都不相同:
--command-config
--producer.config
--consumer.config
意外发现
排查这个问题的时候,我发现了一个颠覆之前对 Kafka 认证连接的认知。之前以为 Kafka 使用 Java API 连接且设置了认证信息后,只有在认证成功后才会出现一句登录成功的日志信息,之前一直以此作为判断 Kafka 有认证场景下成功连接的依据的。
AbstractLogin 类的 login 方法 53 行打印的:
Successfully logged in.
一直对这句话迷惑了,以为只有连接认证信息正确才会打印这句话呢,所以就没有怀疑过认证信息。后面试了一下其他正常的 Kafka 连接信息,发现因网络问题,还是同样的超时问题,而且这次输入的 Kafka 认证信息是错误了,但是日志中也打印了这个信息。
在网络正常的环境下测试输入错误的 Kafka 认证信息,也打印这个日志了,但是后面却会出现 Auth Fail 异常:
2024-10-26 06:48:55.431 [http-nio-xx-exec-2] INFO o.a.k.c.s.a.AbstractLogin - [login,53] - Successfully logged in.
2024-10-26 06:48:55.434 [http-nio-xx-exec-2] WARN o.a.k.c.p.ProducerConfig - [logUnused,246] - The configuration 'producer.type' was supplied but isn't a known config.
2024-10-26 06:48:55.435 [http-nio-xx-exec-2] INFO o.a.k.c.u.AppInfoParser - [<init>,109] - Kafka version : 1.0.2
2024-10-26 06:48:55.436 [http-nio-xx-exec-2] INFO o.a.k.c.u.AppInfoParser - [<init>,110] - Kafka commitId : 2a121f7b1d402825
2024-10-26 06:48:55.576 [kafka-producer-network-thread | producer-8] ERROR o.a.k.c.NetworkClient - [error,296] - [Producer clientId=producer-8] Connection to node -1 failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-256
从日志打印的线程来看,logged in 操作是在主线程中,真正的认证失败信息发生在生产者线程发生数据并调用 get()
方法获取同步响应结果的时候。
再输入一个无效的 Kafka 连接地址,照样打印了 Successfully logged in 信息,只是生产者发送时报 Broker 不可达:
2024-10-26 06:51:33.906 [http-nio-xx-exec-8] INFO o.a.k.c.s.a.AbstractLogin - [login,53] - Successfully logged in.
2024-10-26 06:51:33.909 [http-nio-xx-exec-8] WARN o.a.k.c.p.ProducerConfig - [logUnused,246] - The configuration 'producer.type' was supplied but isn't a known config.
2024-10-26 06:51:33.910 [http-nio-xx-exec-8] INFO o.a.k.c.u.AppInfoParser - [<init>,109] - Kafka version : 1.0.2
2024-10-26 06:51:33.910 [http-nio-xx-exec-8] INFO o.a.k.c.u.AppInfoParser - [<init>,110] - Kafka commitId : 2a121f7b1d402825
2024-10-26 06:51:53.930 [kafka-producer-network-thread | producer-9] WARN o.a.k.c.NetworkClient - [warn,241] - [Producer clientId=producer-9] Connection to node -1 could not be established. Broker may not be available.
结论:基本可以确定一个事实,当使用 Kafka 连接配置认证信息时,无论什么情况「1、信息正确+地址正确;2、都错误;3、地址正确+认证Info错误」,这句 「Successfully logged in
」都会打印,没有任何意义。
跟踪源码可知,它只代表执行了 jaas 的 login
方法,实际没有任何认证相关的操作。想起上周整理过阿里开发者规范的日志规范,这句话应该属于无效日志。
有效的日志打印,应该在数据发送时认证的地方地方打印登录成功才是合理的。
启示录
Kafka 的 API 只会用,从来没有搞明白过调用流程,尤其它的异步线程 kafka-producer-network-thread
,只管再后台无限制重试,无效的 Kafka 地址和信息和真正的 Broker 异常都报一样的错误。
本文这个问题,试过改超时时间为60秒,还是一样的异常,以为自己排查方向错误了。网络上说的这个问题的原因,都一一排除了。
后来用手动发送异步测试正常后,觉得这个同步请求超时还是网络问题,试着把所有的时间参数都改为300秒后,页面在等待2分钟左右的时候有响应了,数据同步操作成功了。
真的是一个 Bug 找一天!