当前位置：首页 > news >正文

浙江众安建设集团有限公司网站南昌县建设局网站

news 2025/11/5 22:28:35

浙江众安建设集团有限公司网站,南昌县建设局网站,建设工程教育网和环球网校哪个好,大连地区网站建设背景某个应用需要配置 Kafka 集群信息，且需要在验证集群是否可达。基本实现思路是创建一个生产者对象，然后发送一条测试数据，调用 Producer 的 send 方法发送消息后，再调用 get() 方法，即同步发送消息，测…

背景

某个应用需要配置 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.