<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version (please specify below)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">airflow version == <code class="notranslate">2.4.0</code></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="task_instance: taskinstance.TaskInstance = context[&quot;ti&quot;]
task_status = task_instance.current_state()"><pre class="notranslate"><code class="notranslate">task_instance: taskinstance.TaskInstance = context["ti"]
task_status = task_instance.current_state()
</code></pre></div>
<p dir="auto">task_status from the above code is returned as "running" when task is failed. It's strange as I see task itself has been marked as failed and also callback gets triggered after failure, but when trying to read the task instance's state it's not as expected.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">task instance should be correctly set to success/failure when trying to read from the callback.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">make an airflow task raise an exception and have the task instance marked as <code class="notranslate">failed</code> and try to read the task instance state within the callback.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">rhel 7.6</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow==2.4.0<br>
apache-airflow-providers-celery==3.0.0<br>
apache-airflow-providers-common-sql==1.2.0<br>
apache-airflow-providers-ftp==3.1.0<br>
apache-airflow-providers-http==4.0.0<br>
apache-airflow-providers-imap==3.0.0<br>
apache-airflow-providers-postgres==5.2.1<br>
apache-airflow-providers-sqlite==3.2.1</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">standard airflow running with postgres backend</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>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.4.0</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">When a task fails in a DAG, the on_failure_callback <a href="https://github.com/apache/airflow/blob/21c40cd4cc7830bc17a4180d8dd0fa2cc4ed9cea/airflow/models/dag.py#L387">registered while creating the dag</a> is triggered using the context of a <a href="https://github.com/apache/airflow/blob/21c40cd4cc7830bc17a4180d8dd0fa2cc4ed9cea/airflow/models/dag.py#L1296">random task instance</a>.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The expectation is that one of the task instances that caused the dag failure should be used instead of a random task instance.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Run the below dag.</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="import datetime
from airflow.models.dag import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator

def all_bad():
    raise Exception(&quot;I have failed&quot;)

def all_good():
    print(&quot;ALL GOOD&quot;)

def failure_callback_dag(context):
    print(&quot;Inside failure_callback_dag&quot;)
    print(context[&quot;task_instance&quot;])
    print(context[&quot;task&quot;])

with DAG(
        dag_id = &quot;test_dag&quot;,
        schedule_interval=None,
        start_date=datetime.datetime(2021, 1, 1),
        catchup=False,
        on_failure_callback=failure_callback_dag
    ) as dag:

    start = EmptyOperator(
        task_id=&quot;start&quot;
    )
    
    fail = PythonOperator(
        provide_context = True,
        task_id = &quot;fail&quot;,
        python_callable = all_bad
    )

    passs = PythonOperator(
        provide_context = True,
        task_id = &quot;pass&quot;,
        python_callable = all_good
    )

    start &gt;&gt; [passs, fail]
"><pre class="notranslate"><span class="pl-k">import</span> <span class="pl-s1">datetime</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">models</span>.<span class="pl-s1">dag</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">python</span> <span class="pl-k">import</span> <span class="pl-v">PythonOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">empty</span> <span class="pl-k">import</span> <span class="pl-v">EmptyOperator</span>

<span class="pl-k">def</span> <span class="pl-en">all_bad</span>():
    <span class="pl-k">raise</span> <span class="pl-v">Exception</span>(<span class="pl-s">"I have failed"</span>)

<span class="pl-k">def</span> <span class="pl-en">all_good</span>():
    <span class="pl-en">print</span>(<span class="pl-s">"ALL GOOD"</span>)

<span class="pl-k">def</span> <span class="pl-en">failure_callback_dag</span>(<span class="pl-s1">context</span>):
    <span class="pl-en">print</span>(<span class="pl-s">"Inside failure_callback_dag"</span>)
    <span class="pl-en">print</span>(<span class="pl-s1">context</span>[<span class="pl-s">"task_instance"</span>])
    <span class="pl-en">print</span>(<span class="pl-s1">context</span>[<span class="pl-s">"task"</span>])

<span class="pl-k">with</span> <span class="pl-v">DAG</span>(
        <span class="pl-s1">dag_id</span> <span class="pl-c1">=</span> <span class="pl-s">"test_dag"</span>,
        <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,
        <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-s1">datetime</span>.<span class="pl-en">datetime</span>(<span class="pl-c1">2021</span>, <span class="pl-c1">1</span>, <span class="pl-c1">1</span>),
        <span class="pl-s1">catchup</span><span class="pl-c1">=</span><span class="pl-c1">False</span>,
        <span class="pl-s1">on_failure_callback</span><span class="pl-c1">=</span><span class="pl-s1">failure_callback_dag</span>
    ) <span class="pl-k">as</span> <span class="pl-s1">dag</span>:

    <span class="pl-s1">start</span> <span class="pl-c1">=</span> <span class="pl-v">EmptyOperator</span>(
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"start"</span>
    )
    
    <span class="pl-s1">fail</span> <span class="pl-c1">=</span> <span class="pl-v">PythonOperator</span>(
        <span class="pl-s1">provide_context</span> <span class="pl-c1">=</span> <span class="pl-c1">True</span>,
        <span class="pl-s1">task_id</span> <span class="pl-c1">=</span> <span class="pl-s">"fail"</span>,
        <span class="pl-s1">python_callable</span> <span class="pl-c1">=</span> <span class="pl-s1">all_bad</span>
    )

    <span class="pl-s1">passs</span> <span class="pl-c1">=</span> <span class="pl-v">PythonOperator</span>(
        <span class="pl-s1">provide_context</span> <span class="pl-c1">=</span> <span class="pl-c1">True</span>,
        <span class="pl-s1">task_id</span> <span class="pl-c1">=</span> <span class="pl-s">"pass"</span>,
        <span class="pl-s1">python_callable</span> <span class="pl-c1">=</span> <span class="pl-s1">all_good</span>
    )

    <span class="pl-s1">start</span> <span class="pl-c1">&gt;&gt;</span> [<span class="pl-s1">passs</span>, <span class="pl-s1">fail</span>]</pre></div>
<p dir="auto">From the dag processor logs:</p>
<p dir="auto">The context that is passed is from the task instance that has succeeded.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2022-09-28T18:28:14.465+0000] {logging_mixin.py:117} INFO - [2022-09-28T18:28:14.463+0000] {dag.py:1292} INFO - Executing dag callback function: &lt;function failure_callback_dag at 0x7fd17ca18560&gt;
[2022-09-28T18:28:14.943+0000] {logging_mixin.py:117} INFO - Inside failure_callback_dag
[2022-09-28T18:28:14.944+0000] {logging_mixin.py:117} INFO - &lt;TaskInstance: test_dag.pass manual__2022-09-28T18:27:59.612118+00:00 [success]&gt;
[2022-09-28T18:28:14.944+0000] {logging_mixin.py:117} INFO - &lt;Task(PythonOperator): pass&gt;"><pre class="notranslate"><code class="notranslate">[2022-09-28T18:28:14.465+0000] {logging_mixin.py:117} INFO - [2022-09-28T18:28:14.463+0000] {dag.py:1292} INFO - Executing dag callback function: &lt;function failure_callback_dag at 0x7fd17ca18560&gt;
[2022-09-28T18:28:14.943+0000] {logging_mixin.py:117} INFO - Inside failure_callback_dag
[2022-09-28T18:28:14.944+0000] {logging_mixin.py:117} INFO - &lt;TaskInstance: test_dag.pass manual__2022-09-28T18:27:59.612118+00:00 [success]&gt;
[2022-09-28T18:28:14.944+0000] {logging_mixin.py:117} INFO - &lt;Task(PythonOperator): pass&gt;
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 10 (buster)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">Default providers that are present in the official airflow docker image.</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">Not sure if it is an expected behaviour, incase it is it needs to be documented in <a href="https://airflow.apache.org/docs/apache-airflow/2.4.0/logging-monitoring/callbacks.html" rel="nofollow">Callbacks</a>.</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>