如何解决org.apache.kafka.clients.NetworkClient Node -1断开连接问题？

我们遇到了Kafka监听器相关问题，无法访问第三方消息发布端。应用启动后恰好9分钟，Spring Boot应用日志中出现以下错误：

org.apache.kafka.clients.NetworkClient [] => [Consumer clientId=consumer-xx, groupId=xx] Node -1 disconnected.

注意到Kafka启动配置日志里显示connections.max.idle.ms = 540000（即9分钟），但按常理，仅9分钟无消息不应该导致应用异常或监听器停止。

奇怪的是，收到消费者断开连接的错误后，监听器依然能正常接收消息；如果消费者是静默重连，理论上应该在最后一条消息接收完成后再过9分钟再次出现断开提示，但实际并未发生，这个行为不符合预期。

应用Kafka配置 #

kafka:
  ssl:
    key-password: xxx
    keystore-location: ${KEYSTORE_LOCATION:xxx.keystore.jks}
    keystore-password: xxx
    truststore-location: ${TRUSTORE_LOCATION:file:xxx.truststore.jks}
    truststore-password: xxx
  consumer:
    bootstrap-servers: ${KAFKA_ENDPOINT:xxx.com:9092}
    group-id: ${KAFKA_GROUP:xxx}
    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  properties:
    session.timeout.ms: ${KAFKA_SESSION_TIMEOUT:10000}
    security.protocol: SSL
    security.inter.broker.protocol: SSL
    sasl.mechanism: https

运行环境 #

• Spring Boot 2.7
• Java 8

应用启动时的ConsumerConfig日志 #

2025-01-13T11:49:59,842Z INFO  restartedMain kafka.clients.consumer.ConsumerConfig [] => ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = latest
    bootstrap.servers = [xxx.com:9092]
    check.crcs = true
    client.dns.lookup = use_all_dns_ips
    client.id = consumer-xxx-1
    client.rack = 
    connections.max.idle.ms = 540000
    default.api.timeout.ms = 60000
    enable.auto.commit = false
    exclude.internal.topics = true
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    group.id = xxx
    group.instance.id = null
    heartbeat.interval.ms = 3000
    interceptor.classes = []
    internal.leave.group.on.close = true
    internal.throw.on.fetch.stable.offset.unsupported = false
    isolation.level = read_uncommitted
    key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.connect.timeout.ms = null
    sasl.login.read.timeout.ms = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.login.retry.backoff.max.ms = 10000
    sasl.login.retry.backoff.ms = 100
    sasl.mechanism = https
    sasl.oauthbearer.clock.skew.seconds = 30
    sasl.oauthbearer.expected.audience = null
    sasl.oauthbearer.expected.issuer = null
    sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
    sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
    sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
    sasl.oauthbearer.jwks.endpoint.url = null
    sasl.oauthbearer.scope.claim.name = scope
    sasl.oauthbearer.sub.claim.name = sub
    sasl.oauthbearer.token.endpoint.url = null
    security.protocol = SSL
    security.providers = null
    send.buffer.bytes = 131072
    session.timeout.ms = 10000
    socket.connection.setup.timeout.max.ms = 30000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = [hidden]
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.certificate.chain = null
    ssl.keystore.key = null
    ssl.keystore.location = xxxkeystore.jks
    ssl.keystore.password = [hidden]
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.2
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.certificates = null
    ssl.truststore.location = xxxtruststore.jks
    ssl.truststore.password = [hidden]
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

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