ExternalTaskSensor should skip if soft_fail=True and external task in one of the failed_states

[alexbegg]

Apache Airflow version

2.1.4

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

What happened

I ran into a scenario where if I use an ExternalTaskSensor and set it to soft_fail=True but also set it to failed_states=['skipped'] I would expect if the external task skipped then to mark this sensor as skipped, however for the failed_states check in the poke method if it is in one of those states it will explicitly fail with an AirflowException.

Wouldn't it make more sense to skip because of the soft_fail?

What you expected to happen

The ExternalTaskSensor task should skip

How to reproduce

1. Add a DAG with a task that is set to skip, such as this BashOperator task set to skip taken from https://airflow.apache.org/docs/apache-airflow/stable/howto/operator/bash.html#skipping:

this_will_skip = BashOperator(
    task_id='this_will_skip',
    bash_command='echo "hello world"; exit 99;',
    dag=dag,
)

2. Add a second DAG with an ExternalTaskSensor
3. Set that sensor to have external_dag_id be the other DAG and external_task_id be the skipped task in that other DAG and failed_states=['skipped'] and soft_fail=True
4. The ExternalTaskSensor fails instead of skips

Anything else

I don't know what is desirable for most Airflow users:

1. To have soft_fail to only cause skips if the sensor times out? (like it seems to currently do)
2. To have ExternalTaskSensor with soft_fail to skip any time it would otherwise fail, such as the external task being in one of the failed_states?
3. To have some other way for the ExternalTaskSensor to skip if the external task skipped?

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

[alexbegg]

Actually, I guess I'll add to this that if ExternalTaskSensor has soft_fail=True it should maybe skip if the external task is in any of the listed failed_states?