Docker环境下Python容器连接Kafka创建Topic报NoBrokersAvailable错误

问题原因及对应修复步骤 #

• 依赖服务未完全就绪 #

  depends_on只能控制容器启动顺序，无法保证Kafka服务内部初始化完成，应用启动时Kafka还未就绪就会触发连接失败。
  修复方案：在Python应用启动前增加依赖检测逻辑，可使用socket循环检测Kafka端口连通性，示例代码如下：

  import socket
  import time
  
  def wait_for_kafka(host, port, timeout=120):
      start_time = time.time()
      while time.time() - start_time < timeout:
          try:
              with socket.create_connection((host, port), timeout=5):
                  return True
          except (ConnectionRefusedError, socket.timeout):
              time.sleep(3)
      raise TimeoutError("Kafka连接超时")
  
  # 实例化KafkaAdminClient前先调用检测
  wait_for_kafka("kafka", 9092)
  

• Kafka监听配置缺失 #

  当前配置仅配置了KAFKA_ADVERTISED_LISTENERS，缺少KAFKA_LISTENERS配置会导致Kafka仅监听内部回环地址，其他容器无法访问。
  修复方案：修改docker-compose.yml中kafka服务的environment配置，补全监听规则：

  kafka:
    image: confluentinc/cp-kafka:5.5.0
    depends_on:
      - zookeeper
    ports:
      - 9092:9092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      # 新增以下行，指定Kafka监听所有网卡的9092端口
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
  

  修改后执行docker-compose up -d --force-recreate kafka重启Kafka容器生效。

• 客户端API版本不兼容 #

  代码中写死的api_version=(0, 10, 2)对应Kafka 0.10.2版本，和你使用的confluent 5.5.0镜像（对应Kafka 2.5版本）不匹配，会导致版本探测失败。
  修复方案：移除实例化KafkaAdminClient时的api_version参数，让客户端自动探测服务端版本，修改后代码如下：

  self._kafka_admin = KafkaAdminClient(
                           bootstrap_servers=[server],
                           api_version_auto_timeout_ms=120000)
  

• 网络连通性验证 #

  若上述配置都正确仍连接失败，可进入app容器手动验证网络连通性：

  1. 执行docker exec -it app bash进入应用容器
  2. 执行ping kafka验证域名解析是否正常
  3. 执行nc -zv kafka 9092验证9092端口是否可访问
     若解析或端口访问失败，检查两个容器的网络配置，确认都加入了rmoff_kafka网络。

成功连接后创建Topic示例代码 #

from kafka.admin import KafkaAdminClient, NewTopic

# 实例化管理客户端
admin_client = KafkaAdminClient(
    bootstrap_servers="kafka:9092",
    api_version_auto_timeout_ms=120000
)
# 定义要创建的Topic配置
topic = NewTopic(
    name="your_topic_name",
    num_partitions=1, # 分区数可按需调整
    replication_factor=1 # 单节点Kafka只能设为1
)
# 执行创建
admin_client.create_topics(new_topics=[topic], validate_only=False)

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