PySpark 2.4.5消费Kafka主题报TimeoutException错误排查

问题复现 #

Spark 2.4.5版本下使用PySpark读取Kafka主题数据，作业运行抛出如下异常：

org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before the position for partition DVI_COMBINED_SUID_LOG-22 could be determined

已确认Kafka集群整体运行状态正常，除当前Spark作业外，其他消费者均可正常消费对应主题，触发异常的业务代码如下：

df = (
    # connect to kafka brokers
    (
        spark.read.format("kafka")
        .option("kafka.bootstrap.servers", kafka_brokers)
        .option("subscribe", kafka_topic)
        .option("includeHeaders", "true")
        .load()
    )
)

df = df.withColumn(
    "message",
    f.from_json(
        f.col("value").cast("string"),
        kafka_schema,
        {"timestampFormat": "yyyy-MM-dd'T'HH:mm:ss.sss'Z'"},
    )
).drop("value")

df.show()

排查&修复步骤 #

按优先级从高到低逐一验证即可：

• 第一优先级验证全节点网络连通性：Spark拉取Kafka数据的逻辑由Executor节点执行，不是Driver节点，必须到每台运行Executor的节点上，测试与Kafka所有broker节点配置的advertised.listeners地址+端口的连通性。90%以上的同类报错都是因为Kafka配置的advertised.listeners返回了私网域名/不可达IP，Executor节点无法建立连接，最终触发位点获取超时，只在Driver节点测telnet没有任何参考价值。
• 显式补充Kafka消费者超时配置，规避默认60s超时过短的问题：Spark 2.4.5内置Kafka客户端默认请求超时仅60s，集群网络稍有波动就会触发失败，在Kafka读取配置中追加以下参数即可：

  .option("kafka.consumer.pollTimeoutMs", "120000")
  .option("kafka.request.timeout.ms", "180000")
  .option("kafka.session.timeout.ms", "120000")
  # 批量读场景下显式指定从最新位点开始消费，避免遍历全分区历史位点额外耗时
  .option("startingOffsets", "latest")
  

• 校验异常分区的副本状态：登录Kafka集群执行分区查询命令，确认报错的DVI_COMBINED_SUID_LOG-22分区leader副本是否在线、是否在ISR同步列表中，哪怕集群整体状态正常，单个分区leader选举卡住、副本同步滞后也会导致位点查询超时。
• 检查Executor资源配额：如果为Executor分配的CPU核数小于1、内存小于1G，Executor启动后工作线程被GC或其他任务占满，无法及时响应Kafka连接请求也会触发超时，建议单Executor分配2核及以上、2G及以上内存重试。
• 规避版本兼容bug：Spark 2.4.5内置的Kafka客户端版本存在位点获取阶段的死锁bug，如果以上配置调整后仍报错，可以在提交作业时排除Spark自带的旧版Kafka客户端依赖，显式引入2.4.1及以上版本的Kafka客户端包重试。

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