Docker环境下Spark Consumer无法找到Kafka Topic分区问题

核心问题分析 #

本地运行Spark应用能正常消费Kafka数据，但提交后出现TimeoutException: 在60000ms超时前无法确定captions-0分区的位置，本质是Spark应用无法正确与Kafka Broker建立网络连接，或无法获取Topic元数据。本地正常说明代码逻辑无问题，问题集中在容器环境的网络配置或Kafka Broker的监听设置。

针对性解决方案 #

1. 修正Kafka的监听地址配置（关键） #

docker-compose.yml中Kafka服务的KAFKA_ADVERTISED_LISTENERS配置是核心，它决定了Kafka向客户端暴露的实际连接地址，配置错误会导致客户端无法定位Broker。示例正确配置：

services:
  broker:
    image: confluentinc/cp-kafka:latest
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      # 监听所有网卡端口
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:29092
      # 向不同网络环境的客户端暴露对应地址
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

• 容器内的Spark应用（与Kafka同属docker-compose网络）使用broker:9092作为bootstrap地址
• 容器外的Spark应用使用localhost:29092作为bootstrap地址

2. 验证Topic状态与连通性 #

进入Kafka容器，执行命令确认Topic存在且可访问：

# 列出所有Topic
kafka-topics.sh --list --bootstrap-server broker:9092
# 测试消费Topic数据
kafka-console-consumer.sh --bootstrap-server broker:9092 --topic captions --from-beginning

若无法消费，说明Topic本身存在问题（如未创建、无数据），需先排查Topic状态。

3. 调整Spark的Kafka连接参数 #

在Spark代码中增加超时参数，避免因网络延迟导致元数据获取失败：

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker:9092") // 根据实际环境调整地址
  .option("subscribe", "captions")
  .option("kafka.session.timeout.ms", "300000") // 延长会话超时时间
  .option("kafka.connection.timeout.ms", "60000") // 延长连接超时时间
  .load()

4. 确认Spark应用的网络环境 #

• 若Spark应用在容器外提交，需确保宿主机的9092/29092端口未被防火墙拦截，且能通过localhost:29092访问Kafka
• 若Spark应用在容器内运行，需确保与Kafka服务同属一个docker-compose网络，否则无法解析broker域名

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