Docker中Go应用容器连接Kafka容器遇Broker耗尽错误求助

错误信息 #

2023/05/28 09:54:44 failed to init resource, error : kafka: client has run out of available brokers to talk to: dial tcp 172.22.0.5:29092: connect: connection refused broker === kafka:29092

环境配置与代码 #

docker-compose.yml配置 #

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - go-tech-test

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_LISTENERS: PLAINTEXT_INTERNAL://kafka:29092,PLAINTEXT://localhost:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT_INTERNAL://kafka:29092,PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT_INTERNAL:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: true
    networks:
      - go-tech-test

  redis-datastore:
    image: redis:latest
    container_name: redis-datastore
    ports:
      - 6379:6379
    networks:
      - go-tech-test

  go-grpc:
    container_name: go-grpc
    build:
      context: .
      dockerfile: ./docker/go_app.Dockerfile
    ports:
      - 8080:8080
    networks:
      - go-tech-test
  
networks:
  go-tech-test:
    driver: bridge

Go应用连接Kafka的核心代码 #

package main

import (
    "fmt"

    "github.com/Shopify/sarama"
    "github.com/redis/go-redis/v9"
)

type Resource struct {
    RedisConn     *redis.Client
    KafkaProducer sarama.SyncProducer
}

func initResource(cfg Config) (Resource, error) {

    redisConn := redis.NewClient(&redis.Options{
        Addr:     cfg.Redis.Host,
        Password: cfg.Redis.Password,
        DB:       0,
    })

    saramaCfg := sarama.NewConfig()
    saramaCfg.Producer.Return.Successes = true
    saramaCfg.Producer.RequiredAcks = sarama.WaitForAll
    saramaCfg.Producer.Retry.Max = 1

    fmt.Printf("broker === %v \n", cfg.Kafka.Broker)

    kafkaConn, err := sarama.NewSyncProducer([]string{cfg.Kafka.Broker}, saramaCfg)
    if err != nil {
        return Resource{}, err
    }

    return Resource{
        RedisConn:     redisConn,
        KafkaProducer: kafkaConn,
    }, nil
}

应用配置文件 #

grpc_port : "8080"

redis:
  host: "redis-datastore:6379"
  password: ""

kafka:
  broker: kafka:29092

排查与解决步骤 #

1. 修正Kafka监听配置 #

当前KAFKA_LISTENERS绑定了kafka主机名，可能导致容器启动时无法正常解析绑定网卡。修改为监听容器内所有网卡：

kafka:
  environment:
    # ... 其他配置
    KAFKA_LISTENERS: PLAINTEXT_INTERNAL://0.0.0.0:29092,PLAINTEXT://0.0.0.0:9092

2. 添加容器健康检查与启动依赖 #

depends_on仅保证容器启动顺序，不确保服务就绪。给Kafka添加健康检查，让Go应用等Kafka完全就绪后再启动：

kafka:
  # ... 其他配置
  healthcheck:
    test: ["CMD", "kafka-topics", "--list", "--bootstrap-server", "localhost:9092"]
    interval: 10s
    timeout: 5s
    retries: 5

go-grpc:
  # ... 其他配置
  depends_on:
    kafka:
      condition: service_healthy

3. 适配Sarama与Kafka版本 #

Kafka 7.3.0对应内核版本2.8.x，在Sarama配置中指定兼容版本，避免协议不匹配：

saramaCfg := sarama.NewConfig()
saramaCfg.Version = sarama.V2_8_0_0 // 匹配Kafka内核版本
saramaCfg.Producer.Return.Successes = true
saramaCfg.Producer.RequiredAcks = sarama.WaitForAll
saramaCfg.Producer.Retry.Max = 1

4. 验证容器间连通性 #

进入Go应用容器，测试与Kafka的网络连通性：

# ping测试主机名解析
docker exec -it go-grpc ping kafka
# telnet测试端口是否开放
docker exec -it go-grpc telnet kafka 29092

如果ping通但telnet失败，说明Kafka未正常监听29092端口；如果ping不通，检查自定义桥接网络是否正常创建。

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