Docker部署Kafka后Python生产者报KafkaTimeoutError错误求助

问题根因 #

该KafkaTimeoutError报错本质是Kafka生产者无法和Kafka Broker正常建立连接，核心有两处配置错误：

1. 运行在Docker容器内的Python代码配置Kafka地址为localhost:9092，指向的是Python容器自身的9092端口，而非Kafka服务的端口
2. Kafka的KAFKA_ADVERTISED_LISTENERS配置为127.0.0.1:9092，返回给客户端的访问地址是回环地址，容器内客户端无法通过这个地址访问到Kafka Broker

修复步骤 #

1. 调整docker-compose.yml中Kafka的监听配置 #

需要配置两类监听端口，分别对应容器内部互访和宿主机外部访问：

version: "3.9"
services:
  zookeeper:
    image: 'bitnami/zookeeper:latest'
    container_name: zookeeper3
    ports:
      - '2181:2181'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:latest'
    container_name: kakfa3
    ports:
      - '9092:9092'
    environment:
      - KAFKA_BROKER_ID=1
      # 配置两个监听地址，INTERNAL供容器内部访问，EXTERNAL供宿主机访问
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka:9093,EXTERNAL://127.0.0.1:9092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper
  python:
    container_name: python3
    build: .
    ports:
      - '5000:5000'
    depends_on:
      - kafka

2. 修改Python代码中Kafka连接地址 #

Docker Compose启动的服务默认在同一个网桥下，容器互访直接使用服务名作为主机名即可：

from kafka import KafkaProducer
import csv
import json

# 连接地址改为Kafka服务名+内部监听端口9093
producer = KafkaProducer(bootstrap_servers=['kafka:9093'],value_serializer=lambda m: json.dumps(m).encode('utf-8'))

# 声明变量存储json消息
jsonArray =[]    

# 读取csv文件并将数据转为json存入数组
with open('sample.csv', 'r') as file:
    csvReader = csv.DictReader(file)
    for row in csvReader: 
        # 将Python字典加入json数组
        jsonArray.append(row)

# 逐个向topic发送消息
for i in range(len(jsonArray)):
    producer.send('part1',jsonArray[i])
    producer.flush()

3. 生效配置 #

修改完成后执行docker-compose up --build重新启动所有服务即可正常发送消息。

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