Spring Boot应用无法连接KRaft模式Kafka Broker的问题求助

问题 #

构建基于Kafka的Spring Boot应用时遇到连接问题，环境及异常信息如下：

• 使用Confluentinc Docker Compose部署3.4版本KRaft模式Kafka（无ZooKeeper）
• Kafdrop可通过29092端口成功连接Kafka，但Spring Boot应用无法正常建立连接

Docker Compose配置（Kafka与Kafdrop部分） #

kafdrop:
  image: obsidiandynamics/kafdrop
  restart: "no"
  ports:
    - "9000:9000"
  environment:
    KAFKA_BROKERCONNECT: "kafka:29092"
    JVM_OPTS: "-Xms16M -Xmx48M -Xss180K -XX:-TieredCompilation -XX:+UseStringDeduplication -noverify"
  depends_on:
    - "kafka"

kafka:
  image: confluentinc/cp-kafka:latest
  restart: "always"
  ports:
    # - "2181:2181"
    - "9092:9092"
  environment:
    KAFKA_NODE_ID: 1
    KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
    KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
    KAFKA_JMX_PORT: 9101
    KAFKA_JMX_HOSTNAME: localhost
    KAFKA_PROCESS_ROLES: 'broker,controller'
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
    KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://localhost:9092'
    KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
    KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
    CLUSTER_ID: 'NzA4MTg2MzAxNDU4MTFlZT'
    KAFKA_RESTART_ATTEMPTS: "10"
    KAFKA_RESTART_DELAY: "5"

Spring Boot Kafka配置 #

spring:
  kafka:
    bootstrap-servers: ${KAFKA_HOST_NAME:localhost}:9092
    producer:
      bootstrap-servers: ${KAFKA_HOST_NAME:localhost}:9092
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        acks: all
        retries: 10
        retry.backoff.ms: 1000
        max.block.ms: 20000
        delivery.timeout.ms: 30000
    admin:
      properties:
        bootstrap.servers: ${KAFKA_HOST_NAME:localhost}:9092
    consumer:
      bootstrap-servers: ${KAFKA_HOST_NAME:localhost}:9092
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: notification-group-id
      properties:
        acks: all
        retries: 10
        retry.backoff.ms: 1000
        max.block.ms: 20000
        delivery.timeout.ms: 30000
        spring:
          json:
            trusted:
              packages: '*'
        bootstrap.servers: ${KAFKA_HOST_NAME:localhost}:9092

异常日志 #

• Kafka容器日志：

[2023-06-27 05:19:53,305] INFO [MetadataLoader 1] handleSnapshot: generated a metadata delta from a snapshot at offset 767 in 12 us. (org.apache.kafka.image.loader.MetadataLoader)

• Spring Boot应用日志：

INFO [,,] 45960 --- [| adminclient-1] org.apache.kafka.clients.NetworkClient   : [AdminClient clientId=adminclient-1] Cancelled in-flight API_VERSIONS request with correlation id 7 due to node -1 being disconnected (elapsed time since creation: 1ms, elapsed time since send: 1ms, request timeout: 3600000ms)

排查与解决方案 #

1. 确认Kafka容器完全就绪 #

KRaft模式下Kafka启动需要完成元数据初始化与选举，MetadataLoader日志仅说明元数据正在加载，需验证服务是否可用：

docker exec -it <kafka-container-name> kafka-topics --list --bootstrap-server localhost:9092

若命令能正常执行（即使无返回主题），说明Kafka就绪；若报错，等待3-5分钟后重试。

2. 简化Spring Boot Kafka配置 #

配置中存在冗余的bootstrap-servers定义，且消费者配置了生产者专属的acks参数，简化后减少潜在冲突：

spring:
  kafka:
    bootstrap-servers: ${KAFKA_HOST_NAME:localhost}:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        acks: all
        retries: 10
        retry.backoff.ms: 1000
        max.block.ms: 20000
        delivery.timeout.ms: 30000
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: notification-group-id
      properties:
        retry.backoff.ms: 1000
        max.block.ms: 20000
        spring:
          json:
            trusted:
              packages: '*'

3. 验证宿主机与Kafka端口的连通性 #

在Spring Boot应用所在机器执行以下命令，检查端口映射是否生效：

nc -zv localhost 9092

若无法连通：

• 执行docker ps确认Kafka容器的9092端口映射为0.0.0.0:9092->9092/tcp
• 重启Docker服务修复端口映射异常

4. 适配Spring Boot的运行环境 #

• 若Spring Boot应用运行在宿主机：保持KAFKA_HOST_NAME=localhost，连接9092端口即可
• 若Spring Boot应用运行在Docker容器（同Docker网络）：设置KAFKA_HOST_NAME=kafka，连接端口改为29092

5. 重置KRaft元数据 #

若Kafka首次启动时初始化异常，残留的元数据可能导致连接问题：

# 停止容器并删除关联数据卷
docker-compose down -v
# 重新启动容器
docker-compose up -d

6. 调整客户端超时参数 #

若Kafka启动较慢，临时增大客户端阻塞超时时间：

spring:
  kafka:
    producer:
      properties:
        max.block.ms: 60000
    consumer:
      properties:
        max.block.ms: 60000

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