K8s容器内Python应用向Kafka发消息成功但无法被消费

你的问题核心在于Kafka的**广告监听地址（Advertised Listeners）**配置错误，导致K8s容器内的生产者获取到了错误的消息发送地址。

问题原因 #

当Python生产者通过host.minikube.internal:9092连接到Kafka后，Kafka会返回KAFKA_ADVERTISED_LISTENERS中配置的地址给生产者，生产者后续会使用这个地址来发送实际的消息。你当前配置的PLAINTEXT_HOST://localhost:9092，对K8s容器来说，localhost指的是容器自身的网络空间，而非运行Kafka的宿主机，这就导致生产者尝试往自己容器的9092端口发送消息，自然无法被Kafka集群接收。

解决方案 #

修改Docker Compose中Kafka的KAFKA_ADVERTISED_LISTENERS配置，将PLAINTEXT_HOST的地址改为host.minikube.internal:9092，让K8s容器能正确解析到Kafka所在的宿主机：

KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://host.minikube.internal:9092

修改完成后，重启Kafka容器：

docker-compose restart broker

额外验证步骤 #

为了确保Python生产者确实成功发送消息，建议在代码中添加Future结果的获取，捕获可能的发送异常：

import kafka

producer = kafka.KafkaProducer(
   bootstrap_servers="host.minikube.internal:9092",
   api_version=(0, 10, 2)
)

try:
    resp = producer.send("events", key=b'message-key', value=b'message-value')
    # 等待发送结果，确认无异常
    record_metadata = resp.get(timeout=10)
    print(f"消息发送成功，分区：{record_metadata.partition}，偏移量：{record_metadata.offset}")
except Exception as e:
    print(f"消息发送失败：{str(e)}")
producer.flush()
producer.close()

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