应用与Kafka集群断开连接：排查及日志配置咨询

问题背景 #

应用与Kafka集群断开连接，应用日志出现如下警告：

14:15:24.445   [kafka-producer-network-thread | producer-1] WARN  o.a.k.c.NetworkClient - [Producer clientId=producer-1] Bootstrap broker A.B.C.D:9093 (id: -2 rack: null) disconnected

但Kafka集群侧无相关日志记录。

当前Kafka集群配置 #

log4j.properties（过滤注释后）

log4j.rootLogger=INFO, kafkaAppender

log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.kafkaAppender=org.apache.log4j.RollingFileAppender
log4j.appender.kafkaAppender.MaxFileSize=50MB
log4j.appender.kafkaAppender.MaxBackupIndex=4
log4j.appender.kafkaAppender.File=/var/log/kafka/kafka-server.log
log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.stateChangeAppender=org.apache.log4j.RollingFileAppender
log4j.appender.stateChangeAppender.MaxFileSize=50MB
log4j.appender.stateChangeAppender.MaxBackupIndex=4
log4j.appender.stateChangeAppender.File=/var/log/kafka/state-change.log
log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.appender.controllerAppender=org.apache.log4j.RollingFileAppender
log4j.appender.controllerAppender.MaxFileSize=50MB
log4j.appender.controllerAppender.MaxBackupIndex=4
log4j.appender.controllerAppender.File=/var/log/kafka/controller.log
log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.logger.kafka=WARN, kafkaAppender

log4j.appender.requestAppender=org.apache.log4j.RollingFileAppender
log4j.appender.requestAppender.MaxFileSize=50MB
log4j.appender.requestAppender.MaxBackupIndex=4
log4j.appender.requestAppender.File=/var/log/kafka/kafka-request.log
log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

log4j.logger.kafka.controller=DEBUG, controllerAppender
log4j.additivity.kafka.controller=false

log4j.logger.state.change.logger=DEBUG, stateChangeAppender
log4j.additivity.state.change.logger=false

server.properties

broker.id=2
delete.topic.enable=true
auto.create.topics.enable=false
listeners=INSIDE://:9091,OUTSIDE1://:9093,OUTSIDE2://:9092
advertised.listeners=INSIDE://kafka-nX:9091,OUTSIDE1://kafka-nX:9093,OUTSIDE2://kafka-nX:9092
listener.security.protocol.map=INSIDE:SASL_SSL,OUTSIDE1:SASL_SSL,OUTSIDE2:PLAINTEXT
inter.broker.listener.name=INSIDE
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
ssl.enabled.protocols=TLSv1.2,TLSv1.1,TLSv1
ssl.secure.random.implementation=SHA1PRNG
super.users=User:admin
ssl.client.auth=none
ssl.keystore.location=/etc/pki/CA/certs/kafka.jks
ssl.keystore.password=password
ssl.key.password=kafkakafka
ssl.truststore.location=/etc/pki/CA/certs/truststore.p12
ssl.truststore.password=password
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
log.dirs=/kafka_data
num.partitions=1
num.recovery.threads.per.data.dir=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.retention.bytes=1073741824
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=A.B.C.D:2181,A.B.C.D+1:2181,A.B.C.D+2:2181
zookeeper.connection.timeout.ms=6000

版本信息 #

• Kafka集群版本：2.8.1
• Kafka客户端版本：3.0.1

问题解答 #

1. 如何监控与分析此类应用断开连接的情况？ #

• 客户端侧：
  • 采集日志中的disconnected警告，统计出现频率、关联broker地址和业务时间点，判断是否和业务峰值、网络波动同步。
  • 临时开启org.apache.kafka.clients.NetworkClient的DEBUG日志，查看断开前的网络交互细节，比如请求超时、SSL握手失败等前置异常。
  • 监控客户端内置指标：比如producer的producer_network_io_disconnects_total、consumer的consumer_network_io_disconnects_total，用Prometheus+Grafana可视化并设置阈值告警。
• 集群侧：
  • 监控broker的kafka_network_connections_current指标，观察连接数是否有异常波动，对应客户端断开时间点。
  • 查看kafka.server:type=SessionExpireListener,name=ExpiredSessionsPerSec指标，统计会话过期频率，判断是否为会话超时导致断开。
• 网络层面：
  • 在客户端与broker之间抓包，分析TCP断开原因（FIN/RST包），排查防火墙、负载均衡的超时策略是否主动断开连接。
  • 用ping、mtr等工具检测链路延迟、丢包率，排查网络稳定性。

2. 如何配置Kafka集群侧的log4j.properties文件以查看此类断开会话？ #

要捕获broker侧的连接断开日志，需调整以下日志配置：

• 添加网络模块的DEBUG日志，捕获连接断开事件：

  log4j.logger.org.apache.kafka.common.network=DEBUG, kafkaAppender
  log4j.additivity.org.apache.kafka.common.network=false
  

• 若使用SSL连接，补充开启安全模块的DEBUG日志，排查SSL层面问题：

  log4j.logger.org.apache.kafka.common.security=DEBUG, kafkaAppender
  log4j.additivity.org.apache.kafka.common.security=false
  

• 配置后，断开日志会输出到kafka-server.log，示例如下：

  [2024-05-20 14:15:24,445] DEBUG [SocketServer brokerId=2] Connection with /X.X.X.X closed (org.apache.kafka.common.network.Selector)
  

• 注意：DEBUG级别会增加日志量，排查完成后建议调回原有级别，避免磁盘占用过高。

3. 可能导致该问题的原因有哪些？ #

• 网络层面：
  • 客户端与broker链路不稳定，出现丢包、高延迟，导致TCP连接超时断开。
  • 防火墙、负载均衡设置了连接超时，长时间无数据交互时主动断开连接。
  • 带宽不足，业务峰值时数据包堆积引发连接断开。
• 配置与版本问题：
  • 客户端与集群版本不匹配（3.0.1 vs 2.8.1），存在协议兼容性问题导致连接异常。
  • broker的socket.connection.max.idle.ms设置过小，客户端空闲超时被主动断开。
  • SSL配置不一致：客户端与broker的SSL协议、加密套件不匹配，引发握手失败或中途断开。
• 资源限制：
  • broker文件描述符不足，无法维持现有连接导致强制断开。
  • broker CPU、内存耗尽，无法及时处理客户端请求引发连接超时。
• 客户端自身问题：
  • 客户端进程重启、资源耗尽，主动断开连接。
  • 客户端metadata.max.age.ms、reconnect.backoff.ms等参数不合理，导致元数据更新不及时或重连策略失效。

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