如何修复Cloud Composer中‘Task Instance Not Running’依赖失败问题？

问题日志 #

[2022-10-03, 00:34:03 UTC] {taskinstance.py:1034} INFO - Dependencies not met for <TaskInstance: dag_name.task_name scheduled__2022-10-02T00:00:00+00:00 [running]>, dependency 'Task Instance Not Running' FAILED: Task is in the running state
[2022-10-03, 00:34:03 UTC] {taskinstance.py:1034} INFO - Dependencies not met for <TaskInstance: dag_name.task_name scheduled__2022-10-02T00:00:00+00:00 [running]>, dependency 'Task Instance State' FAILED: Task is in the 'running' state.

该问题在传感器任务、普通任务中均会出现，甚至谷歌官方提供的airflow_monitoring DAG也存在此问题。任务会因此卡住，必须手动重启才能恢复，可通过修改配置解决。

环境配置 #

Google Cloud Composer 1.19.1
Airflow 2.2.5
运行于Kubernetes集群，采用Celery executor

可修改的配置项及操作 #

1. 调整任务超时与状态追踪设置 #

在Composer环境的Airflow配置中修改以下参数：

• core.dag_run_conf_overrides_params: 设置为True，允许DAG运行时配置覆盖默认参数
• celery.task_track_started: 设置为True，确保Celery任务的启动状态被正确追踪，避免状态误判
• core.task_instance_timeout: 根据任务实际耗时调整，例如设为86400（单位：秒，即24小时），防止任务长时间无响应却仍处于"running"状态

2. 启用僵尸任务自动清理 #

开启调度器的僵尸任务检测与清理机制：

• scheduler.zombie_task_threshold: 设置为300（单位：秒，即5分钟），调度器会自动标记超过该时长无心跳的任务为失败，触发重新调度
• scheduler.num_runs: 保持默认或适当调高，确保调度器能频繁检查任务状态

3. 传感器任务专属优化 #

针对传感器任务，可通过全局配置或DAG定义调整：

• sensor_mode: 改为reschedule模式（替代默认的poke模式），避免传感器长时间占用Worker资源导致状态异常
• core.default_task_retries: 适当调高重试次数，例如设为3，让任务自动重试而非卡住

操作方式 #

在Google Cloud Composer控制台进入对应环境的「Airflow配置」页面，找到上述参数修改后保存，配置会自动生效，无需重启集群。

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