Python Kafka Producer连接异常但Consumer可正常连接问题排查

本地通过docker-compose部署Kafka与ZooKeeper集群后，出现以下异常：KafkaProducer初始化无报错，但调用producer.send()时触发连接拒绝（ECONNREFUSED）、地址不可用（EADDRNOTAVAIL）等KafkaConnectionError；而使用相同bootstrap_servers配置的KafkaConsumer却能正常连接。

相关代码与配置 #

生产者代码 #

producer = KafkaProducer(bootstrap_servers=SERVER, security_protocol="PLAINTEXT") # 初始化无报错，若无法连接Broker会抛出NoBrokersAvailable错误
producer.send("g", "test") 

消费者代码 #

consumer = KafkaConsumer(
    "g",
    group_id="g",
    bootstrap_servers=SERVER,
)

docker-compose配置 #

zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
        ZOOKEEPER_CLIENT_PORT: 2181
        ZOOKEEPER_TICK_TIME: 2000
    ports:
        - 22181:2181

kafka:
    image: confluentinc/cp-kafka:5.3.1
    depends_on:
        - zookeeper
    ports:
        - 29092:29092
    environment:
        KAFKA_BROKER_ID: 1
        KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
        KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
        KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
        KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
        KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

错误日志 #

decoding-server  | [E 221216 16:23:01 conn:418] Connect attempt to <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv4 ('127.0.0.1', 29092)]> returned error 111. Disconnecting.
decoding-server  | [I 221216 16:23:01 conn:919] <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv4 ('127.0.0.1', 29092)]>: Closing connection. KafkaConnectionError: 111 ECONNREFUSED
decoding-server  | [W 221216 16:23:01 client_async:331] Node 1 connection failed -- refreshing metadata
decoding-server  | [I 221216 16:23:02 conn:380] <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv6 ('::1', 29092, 0, 0)]>: connecting to localhost:29092 [('::1', 29092, 0, 0) IPv6]
decoding-server  | [E 221216 16:23:02 conn:418] Connect attempt to <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv6 ('::1', 29092, 0, 0)]> returned error 99. Disconnecting.
decoding-server  | [I 221216 16:23:02 conn:919] <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv6 ('::1', 29092, 0, 0)]>: Closing connection. KafkaConnectionError: 99 EADDRNOTAVAIL
decoding-server  | [W 221216 16:23:02 client_async:331] Node 1 connection failed -- refreshing metadata
decoding-server  | [I 221216 16:23:03 conn:380] <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv4 ('127.0.0.1', 29092)]>: connecting to localhost:29092 [('127.0.0.1', 29092) IPv4]
decoding-server  | [E 221216 16:23:03 conn:418] Connect attempt to <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv4 ('127.0.0.1', 29092)]> returned error 111. Disconnecting.
decoding-server  | [I 221216 16:23:03 conn:919] <BrokerConnection node_id=1 host=localhost:29092 <connecting> [IPv4 ('127.0.0.1', 29092)]>: Closing connection. KafkaConnectionError: 111 ECONNREFUSED
decoding-server  | [W 221216 16:23:03 client_async:331] Node 1 connection failed -- refreshing metadata

解决方案 #

问题根源在于Kafka对外宣告的监听地址与客户端网络环境不匹配：

• 消费者初始化仅需通过bootstrap地址获取元数据，而生产者发送消息时需要直接连接Broker宣告的实际地址。当前配置中Kafka的PLAINTEXT_HOST宣告地址为localhost:29092，若生产者运行在Docker容器（如decoding-server）内，容器内的localhost指向自身，而非宿主机，因此无法连接到Kafka服务。

修复步骤 #

1. 调整Kafka的KAFKA_ADVERTISED_LISTENERS配置：

   • 若生产者在宿主机运行：将PLAINTEXT_HOST地址改为宿主机IP

     KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://你的宿主机IP:29092
     

   • 若生产者与Kafka在同一Docker网络：直接使用Kafka容器名作为地址

     KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://kafka:9092
     

2. 同步调整生产者的bootstrap_servers：

   • 宿主机运行：SERVER = "localhost:29092"
   • Docker容器同网络运行：SERVER = "kafka:9092"

关键说明 #

Kafka的advertised.listeners是Broker对外暴露的实际连接地址，客户端获取元数据后会使用该地址建立连接，而非初始化时的bootstrap地址，这是生产者与消费者表现差异的核心原因。

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