Airflow自定义Operator触发执行超时后任务未停止问题排查

嘿，我来帮你揪出这个超时触发但任务还在跑的问题！结合Airflow的运行机制，你大概率是遗漏了这几个关键环节：

1. 确保execution_timeout正确传递给父类 #

首先得确认自定义Operator的__init__方法里，把execution_timeout参数传给了BaseOperator的构造函数。如果父类没拿到这个参数，Airflow的超时监控逻辑根本不会启动。

比如你的代码应该调整成这样：

from airflow.models.baseoperator import BaseOperator
from airflow.utils.decorators import apply_defaults

class MyOperator(BaseOperator):
    @apply_defaults
    def __init__(
        self,
        some_parameters_here,
        execution_timeout=None,  # 显式声明超时参数
        *args, **kwargs
    ):
        # 必须把execution_timeout传给父类初始化
        super().__init__(execution_timeout=execution_timeout, *args, **kwargs)
        # 你的其他自定义初始化逻辑

要是之前没做这一步，哪怕你实例化Operator时传了execution_timeout=timedelta(minutes=30)，父类也接收不到，自然不会触发超时终止。

2. 在execute方法中定期检查终止信号 #

Airflow的超时不是实时强制终止的——调度器检测到超时后，会给Worker发终止信号，但如果你的execute方法里是长时间阻塞的操作（比如死循环、无超时的数据库查询），而且没有主动检查任务状态，那任务不会主动停止。

解决方法是在execute的循环或关键步骤中，调用BaseOperator提供的check_for_termination()方法，它会检查任务是否被标记为终止，是的话就抛出异常终止任务：

def execute(self, context):
    while your_running_condition:
        # 执行你的任务逻辑
        # 定期检查是否需要终止，比如每10秒检查一次
        self.check_for_termination()
        time.sleep(10)

这样即使是长时间运行的任务，也能及时响应超时信号。

3. 处理子进程的终止信号 #

如果你的execute方法里启动了子进程（比如用subprocess调用外部脚本），那即使主进程被Airflow终止，子进程可能还会继续运行，导致任务看起来“还在跑”。

这种情况下，你需要在启动子进程时确保它能接收终止信号，或者在主进程终止时主动杀掉子进程。比如用subprocess.Popen启动子进程，然后在check_for_termination抛出异常时，调用p.kill()：

def execute(self, context):
    p = subprocess.Popen(["your_script.sh"])
    try:
        while p.poll() is None:
            self.check_for_termination()
            time.sleep(5)
    except Exception as e:
        # 任务被终止时，杀掉子进程
        p.kill()
        raise e

4. 确认Airflow Worker的信号处理能力 #

如果用的是CeleryExecutor，还要确保Worker能正确接收SIGTERM/SIGINT信号。有些环境下（比如容器化部署），Worker可能因为进程PID的问题，无法收到调度器发的终止信号，导致超时后任务继续运行。这种情况可以检查Worker的日志，看是否有接收终止信号的记录。

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