Airflow 2.0.1 PythonOperator任务无论是否抛异常均返回成功状态问题排查

问题根源 #

你遇到的PythonOperator抛出异常仍标记为成功的问题，通常由以下3种原因导致：

1. 异常被静默捕获：main函数内部或其调用的子方法中存在全局try-except块，捕获了包括AirflowException在内的所有异常但未重新抛出，异常无法向上传递到PythonOperator的执行器，因此执行器判定任务执行成功。
2. 多执行流异常未传递：如果main函数中启动了子线程/子进程执行业务逻辑，子执行流抛出的异常默认不会传递到主线程，主线程正常执行完成就会返回成功状态。
3. 版本已知bug：Airflow 2.0.1属于早期2.x迭代版本，存在PythonOperator异常处理的已知问题，部分场景下抛出的异常会被执行器忽略。

排查&解决步骤 #

• 最简复现验证
  先将main函数替换为无业务逻辑的测试代码，确认是否能复现问题：

  from airflow.exceptions import AirflowException
  def main():
      raise AirflowException("测试异常")
  

  如果替换后任务依然标记为成功，排除业务代码吞异常的问题，继续后续排查。如果替换后任务正常标记为失败，说明你的原业务代码中存在异常捕获逻辑，找到对应位置去掉全局捕获或者捕获后重新抛出即可。
• 确认导入函数正确性
  在DAG文件导入main函数后增加一行打印，确认导入的是目标文件的函数：

  from utils.main import main
  print("导入的main函数路径：", main.__code__.co_filename)
  

  查看任务运行日志中输出的路径，确认和你编写的path_to_airflow/dags/utils/main.py路径一致，避免导入了其他位置的同名函数。
• 多执行流异常适配
  如果你用到了多线程/多进程，需要手动捕获子执行流的异常并在主线程重新抛出，多线程捕获示例如下：

  import threading
  class ExceptionThread(threading.Thread):
      def run(self):
          self.exception = None
          try:
              super().run()
          except Exception as e:
              self.exception = e
      def join(self, *args, **kwargs):
          super().join(*args, **kwargs)
          if self.exception:
              raise self.exception
  

• 版本升级
  若确认是Airflow 2.0.1的版本bug，建议升级到2.2及以上的LTS稳定版本，异常处理逻辑已经修复。
• 临时替代方案
  如果不想修改现有逻辑，可以保持使用BashOperator直接调用Python脚本执行，异常会通过进程退出码被Airflow正常捕获。

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