Deferrable ExternalTaskSensor cannot wait for an entire DAG

[ecodina]

Apache Airflow version

2.7.1

What happened

When running an ExternalTaskSensor with external_task_id=None and in deferrable mode, the trigger doesn't wait for the entire DAG since it needs a task_id. However, the typing suggests this should be possible: https://github.com/apache/airflow/blob/main/airflow/triggers/external_task.py#L63

I believe this is the query that doesn't behave as expected from the typing: https://github.com/apache/airflow/blob/main/airflow/triggers/external_task.py#L135

Moreover, after examining the code, it seems the trigger would also fail if the ExternalTaskSensor was provided with an external_task_ids (multiple tasks).

What you think should happen instead

No response

How to reproduce

1. Create dag1 which contains a simple task (e.g. sleep for 60 seconds)
2. Create dag2 which contains an ExternalTaskSensor checking for dag1 (external_task_id=None) and runs in deferrable mode
3. Run them and see how dag2 times out because it can't find any task.

Operating System

Ubuntu 22.04.3 LTS

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

Anything else

Similarly to #34204 I'd love to submit a PR, but have no available time to properly do so or the knowledge to make sure any of my solutions actually work in all cases.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[samc1213]

I'm interested in taking this issue

[samc1213]

I think this file causes the issue? I see dag2 in status failed - is this a valid way to reproduce the issue @ecodina ? I'm new to airflow so thanks for your help. How do you know it's timing out instead of just failing?

[ecodina]

Yes, this file reproduces the issue. You can try lowering the sleep time (e.g. 10 seconds). You'll then see that the ExternalTask keeps logging Task is still running, sleeping for 2.0 seconds... after the sleeping task has long finished. It fails after the trigger's 60 seconds hardcoded timeout.

You can also print the result of TaskStateTrigger.count_tasks and check it's always 0.

I opened 3 issues regarding the deferrable ExternalTaskSensor and as @yermalov-here commented in #34204 the problem is that the logic of this deferrable trigger is different from the "normal" one. Looking in retrospective, maybe it'd have been better to open 1 issue with the 3 "bugs" since a total refactor may be needed!

[samc1213]

Should we consider consolidating the issues? The conversation in #34204 is much more thorough and the fix should address this problem too

[ecodina]

Sorry, OOO until October and I just saw your message. It's probably a good idea to consolidate all the issues! I think they all share the same root-cause.