Repository navigation
TaskFlow AirflowSkipException causes downstream step to fail #24338
Description
Apache Airflow version
2.3.2 (latest released)
What happened
Using TaskFlow API and have 2 tasks that lead to the same downstream task. These tasks check for new data and when found will set an XCom entry of the new filename for the downstream to handle. If no data is found the upstream tasks raise a skip exception.
The downstream task has the trigger_rule = none_failed_min_one_success.
Problem is that a task which is set to Skip doesn't set any XCom. When the downstream task starts it raises the error:
airflow.exceptions.AirflowException: XComArg result from task2 at airflow_2_3_xcomarg_render_error with key="return_value" is not found!
What you think should happen instead
Based on trigger rule of "none_failed_min_one_success", expectation is that an upstream task should be allowed to skip and the downstream task will still run. While the downstream does try to start based on trigger rules, it never really gets to run since the error is raised when rendering the arguments.
How to reproduce
Example dag will generate the error if run.
from airflow.decorators import dag, task
from airflow.exceptions import AirflowSkipException
@task
def task1():
return "example.csv"
@task
def task2():
raise AirflowSkipException()
@task(trigger_rule="none_failed_min_one_success")
def downstream_task(t1, t2):
print("task ran")
@dag(
default_args={"owner": "Airflow", "start_date": "2022-06-07"},
schedule_interval=None,
)
def airflow_2_3_xcomarg_render_error():
t1 = task1()
t2 = task2()
downstream_task(t1, t2)
example_dag = airflow_2_3_xcomarg_render_error()
Operating System
Ubuntu 20.04.4 LTS
Versions of Apache Airflow Providers
No response
Deployment
Virtualenv installation
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
- I agree to follow this project's Code of Conduct
Activity
robinsinghstudios commentedon Jun 10, 2022
The same issue has been bothering me for a while now and preventing me from upgrading to the latest airflow version.
vanchaxy commentedon Jun 11, 2022
As a workaround, you can remove t2 from downstream_task args and check if task2 result exists in the function body. Or you can add one more task between task2 and downstream_task that will check if upstream skipped and return None or return upstream value.
I think downstream_task should be run in your case, but I'm not sure want value we expect to see in t2? None/skip exception/some_special_value_for_that_case?
mariotaddeucci commentedon Jun 13, 2022
I have the same issue, after upgrading from 2.1.3 to 2.3.2, some dags with the trigger rule all_done, starting to raises this exception
hterik commentedon Jun 23, 2022
Same issue when using the branchoperator, eg
We want to run either t1 or t2, or both, then combine_results the results, but if any of them are skippped we observe the same issue,
XComArg result from .... with key="return_value" is not found!Previously we would get null values as input arguments if the upstream was skipped, that's not ideal either, some placeholder-value for skipped would be the best imo.ashb commentedon Jul 5, 2022
@jordanjeremy What value would you expect to be passed to
downstream_taskfor thet2argument?jordanjeremy commentedon Jul 5, 2022
@ashb I provided additional detail in #24401 (comment) showing a more complete example of how this was being used.
In short, passing the XComArg to a downstream task was being used as it allowed the relationship between tasks to be established automatically instead of having to explicitly set these relationships through bitshift operators or calling set_upstream/set_downstream.
ashb commentedon Jul 5, 2022
@jordanjeremy That says the use case (I agree it makes sense) but not the runtime behaviour. I guess you're expecting
Noneto be passed?jordanjeremy commentedon Jul 5, 2022
@ashb Previously this did return
None. However, I do see that this could be indeterminant for the cases where XCom value could have been set to None or it may not have been set. Would returning the NOTSET object instead of raising an error when it is seen work better? If that is done, then the difference between XCom is None or XCom wasn't set, could be determined in the event that someone needed to be able to tell the difference between those cases.Edit: I think the most important runtime behavior is that the
XComArg.resolvemethod returns a value instead of raising an error. When raising an error the task won't run, even if by the trigger rules meant that it tried to run. The exact value returned, None or some sentinel value (NOTSET) that someone can check for should work, in my opinion.markhatch commentedon Jul 15, 2022
Unsure if helpful - but tossing in my vote for this as well and thought to share my use case.
Expected
trigger_rulewould be respected rather than automatically failing downstream tasks. I have downstream task that pick a random choice from any successful upstream.In the example above - would expect the
choose_clustertask to pick/passcluster_b. Instead throwskey="return_value" is not foundas mentioned.