Issue 1:
Apache Airflow version

2.3.0

What happened

Hello!

I'm using taskflow api for my dag and was trying to run the last task even if the previous task failed, I used @task(trigger_rule=TriggerRule.ALL_DONE) but i still got the ending task failed too.

The UI confirms that I declared the trigger_rule "all_done"

What you think should happen instead

No response

How to reproduce

This is the whole code

import logging
import pendulum
from airflow.decorators import task, dag
from airflow.utils.trigger_rule import TriggerRule

@dag(
schedule_interval=None,
start_date=pendulum.datetime(2021, 5, 10, tz="UTC"),
catchup=False,
)
def taskflow_trigger():
@task()
def first_task():
return "first_task"

@task
def second_task(value):
return "second_task"

@task
def task_to_fail(value):
data = {"test": 1}
val = data["not_here"]
return val

@task(trigger_rule=TriggerRule.ALL_DONE)
def end(value):

return "yes"

op_1 = first_task()
op_3 = task_to_fail(op_1)
op_2 = second_task(op_1)
end([op_3, op_2])

taskflow_trigger = taskflow_trigger()

Operating System

"Debian GNU/Linux 11 (bullseye)

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

I agree to follow this project's Code of Conduct

Issue 2:
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