<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Hello!</p>
<p dir="auto">I'm using taskflow api for my dag and was trying to run the last task even if the previous task failed, I used <code class="notranslate">@task(trigger_rule=TriggerRule.ALL_DONE)</code> but i still got the ending task failed too.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/38496723/170526480-131e6ed4-e2a0-4711-9328-2326f94885e3.png"><img width="213" alt="image" src="https://user-images.githubusercontent.com/38496723/170526480-131e6ed4-e2a0-4711-9328-2326f94885e3.png" style="max-width: 100%;"></a></p>
<p dir="auto">The UI confirms that I declared the trigger_rule "all_done"<br>
<a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/38496723/170526766-72ba0eb7-2525-4d60-a8c3-ab3dd700d88b.png"><img width="218" alt="image" src="https://user-images.githubusercontent.com/38496723/170526766-72ba0eb7-2525-4d60-a8c3-ab3dd700d88b.png" style="max-width: 100%;"></a></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">This is the whole code</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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=&quot;UTC&quot;),
    catchup=False,
)
def taskflow_trigger():
    @task()
    def first_task():
        return &quot;first_task&quot;

    @task
    def second_task(value):
        return &quot;second_task&quot;

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

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

        return &quot;yes&quot;

    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()"><pre class="notranslate"><code class="notranslate">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()
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">"Debian GNU/Linux 11 (bullseye)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Docker-Compose</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.2 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">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.<br>
The downstream task has the trigger_rule = none_failed_min_one_success.</p>
<p dir="auto">Problem is that a task which is set to Skip doesn't set any XCom.  When the downstream task starts it raises the error:<br>
<code class="notranslate">airflow.exceptions.AirflowException: XComArg result from task2 at airflow_2_3_xcomarg_render_error with key="return_value" is not found!</code></p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">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.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Example dag will generate the error if run.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="from airflow.decorators import dag, task
from airflow.exceptions import AirflowSkipException

@task
def task1():
    return &quot;example.csv&quot;

@task
def task2():
    raise AirflowSkipException()

@task(trigger_rule=&quot;none_failed_min_one_success&quot;)
def downstream_task(t1, t2):
    print(&quot;task ran&quot;)

@dag(
    default_args={&quot;owner&quot;: &quot;Airflow&quot;, &quot;start_date&quot;: &quot;2022-06-07&quot;},
    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()"><pre class="notranslate"><code class="notranslate">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()
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 20.04.4 LTS</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Virtualenv installation</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>