Kafka空闲连接识别机制及SpringBoot生产者连接断开问题咨询

问题场景 #

测试SpringBoot Kafka生产者时，设置每30秒发送一次请求，持续1小时。开启TRACE日志后，约10分钟出现连接断开的EOFException，错误日志如下：

2022-07-18 19:15:25.025 DEBUG 12991 --- [streams-kafka-1] o.apache.kafka.common.network.Selector   : [Producer clientId=event-streams-kafka-1] Connection with localhost/127.0.0.1 disconnected

java.io.EOFException: null
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:97) ~[kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452) ~[kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402) ~[kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:576) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.poll(Selector.java:481) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:551) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:328) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:243) [kafka-clients-3.0.0.jar:na]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_311]

2022-07-18 19:15:25.027 TRACE 12991 --- [streams-kafka-1] o.apache.kafka.common.network.Selector   : [Producer clientId=event-streams-kafka-1] Read from closing channel failed, ignoring exception

java.io.EOFException: null
    at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:97) ~[kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452) ~[kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402) ~[kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.maybeReadFromClosingChannel(Selector.java:700) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.close(Selector.java:935) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:625) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.common.network.Selector.poll(Selector.java:481) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:551) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:328) [kafka-clients-3.0.0.jar:na]
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:243) [kafka-clients-3.0.0.jar:na]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_311]

2022-07-18 19:15:25.027 DEBUG 12991 --- [streams-kafka-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=event-streams-kafka-1] Node -2 disconnected.

尝试在客户端设置connections.max.idle.ms=-1（永不主动关闭空闲连接）无效，但修改Broker端的connections.max.idle.ms后问题消失。

核心疑问解答 #

1. Kafka如何判定连接为空闲状态？ #

Kafka Broker通过connections.max.idle.ms参数（默认值540000ms=9分钟）控制空闲连接超时，判定逻辑：

• 从连接上最后一次完整的请求/响应交互结束开始计时，期间如果没有任何新的请求（消息发送、元数据查询、控制类请求等）或响应交互，累计时长达到阈值后，Broker会主动断开该连接。
• 注意：客户端的connections.max.idle.ms是客户端主动关闭自身空闲连接的阈值，和Broker主动断开是两个独立逻辑，所以设置客户端参数为-1无法阻止Broker的主动断开行为。

2. 每30秒发送请求为何还会被判定空闲？ #

虽然每30秒发送一次消息，但可能存在以下情况：

• 每次消息发送的响应完成后，到下一次发送的30秒窗口内，没有任何其他交互（比如客户端未主动发起元数据查询），Broker的空闲计时器持续累计，直到接近默认9分钟阈值时触发断开。
• Kafka生产者默认metadata.max.age.ms为300000ms=5分钟，如果Topic元数据无变化，客户端不会主动发送元数据请求，期间只有每30秒的消息发送，一旦某次消息响应结束后，累计空闲时间叠加到9分钟，就会触发Broker断开。

3. 客户端层面如何避免连接被关闭？ #

无需修改Broker全局参数，调整客户端配置即可解决：

• 调整metadata.max.age.ms参数：将其设置为小于Brokerconnections.max.idle.ms的值（比如240000ms=4分钟），让客户端定期主动发送元数据查询请求，持续保持连接活跃，避免Broker判定为空闲。
• 检查连接复用配置：确保生产者的max.in.flight.requests.per.connection（默认5）、batch.size等配置合理，避免连接频繁新建/销毁，保证连接持续复用。

4. 关于Broker参数修改的顾虑 #

如果必须调整Broker参数，建议：

• 将connections.max.idle.ms调整到更合理的全局值（比如1800000ms=30分钟），既满足长连接需求，也不会对其他应用造成明显影响。
• Kafka目前不支持针对特定客户端或Topic设置该参数，全局调整是唯一可行的Broker端方案，但优先推荐客户端层面的配置调整。

内容的提问来源于stack exchange，提问作者chewie