ExternalTaskSensor doesn't check existence in deferrable model

apache / airflow

Apache Airflow - A platform to programmatically author, schedule, and monitor workflows

https://airflow.apache.org/

Apache License 2.0

36.71k stars 14.21k forks source link

ExternalTaskSensor doesn't check existence in deferrable model #40745

Open armandduijn opened 3 months ago

armandduijn commented 3 months ago

Apache Airflow version

2.9.2

If "Other Airflow 2 version" selected, which one?

No response

What happened?

The ExternalTaskSensor in deferrable mode doesn't throw an error if the external DAG doesn't exist. Setting check_existence=True doesn't have an effect.

What you think should happen instead?

The task should immediately fail instead of staying in the deferred state.

How to reproduce

The code below implements a DAG with two ExternalTaskSensor tasks. One task is set to be deferred and the other one is old non-deferred way.

When running the DAG, the regular_sensor task will immediately retry and eventually fail, while the defer_sensor task will remain deferred.

from datetime import datetime

from airflow import DAG
from airflow.sensors.external_task import ExternalTaskSensor

with DAG(
    "Test",
    default_args={ "start_date": datetime(2024, 7, 11) }
) as dag:
    ExternalTaskSensor(
        task_id="defer_sensor",
        external_dag_id="FakeDAG",
        check_existence=True,
        deferrable=True,
    )

    ExternalTaskSensor(
        task_id="regular_sensor",
        external_dag_id="FakeDAG",
        check_existence=True,
        deferrable=False,
    )

Operating System

Windows

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

[ ] Yes I am willing to submit a PR!

Code of Conduct

[X] I agree to follow this project's Code of Conduct

Prab-27 commented 3 months ago

hi @eladkal , Would you please assign this issue .I want to contribute as my first contribution and I assure you I will complete as soon as possible.

potiuk commented 3 months ago

Assigned

vikramaditya91 commented 3 months ago

I was planning on working on this but happy to see that @Prab-27 is going to be contributing towards this.

In any case, my original plan was to fix this by following:

Move the _check_for_existence method from the ExternalTaskSensor to a shared location (potentially in the airflow/airflow/utils/) and call it the check_for_existence method. Make changes in the ExternalTaskSensor, because of this change
Add an extra argument to the constructor of the WorkflowTrigger with a new keyword argument check_existence. It is False by default, but is being set by the check_existence in ExternalTaskSensor.
In the async run of the WorkflowTrigger, call the check_for_existence if check_existence is True. This step will raise an AirflowException is the DAG/task_id/task_group_id is not found

Do you agree with that solution @potiuk ?

potiuk commented 3 months ago

Do you agree with that solution @potiuk ?

As usual - seing the PR is best - we can discuss it there, it's usually better than discussing it "dry" :)