Issue 1:
Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

airflow version == 2.4.0

task_instance: taskinstance.TaskInstance = context["ti"]
task_status = task_instance.current_state()

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.

What you think should happen instead

task instance should be correctly set to success/failure when trying to read from the callback.

How to reproduce

make an airflow task raise an exception and have the task instance marked as failed and try to read the task instance state within the callback.

Operating System

rhel 7.6

Versions of Apache Airflow Providers

apache-airflow==2.4.0

apache-airflow-providers-celery==3.0.0

apache-airflow-providers-common-sql==1.2.0

apache-airflow-providers-ftp==3.1.0

apache-airflow-providers-http==4.0.0

apache-airflow-providers-imap==3.0.0

apache-airflow-providers-postgres==5.2.1

apache-airflow-providers-sqlite==3.2.1

Deployment

Other

Deployment details

standard airflow running with postgres backend

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.4.0

What happened

When a task fails in a DAG, the on_failure_callback registered while creating the dag is triggered using the context of a random task instance.

What you think should happen instead

The expectation is that one of the task instances that caused the dag failure should be used instead of a random task instance.

How to reproduce

Run the below dag.

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("I have failed")

def all_good():
print("ALL GOOD")

def failure_callback_dag(context):
print("Inside failure_callback_dag")
print(context["task_instance"])
print(context["task"])

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

start = EmptyOperator(
task_id="start"
)

fail = PythonOperator(
provide_context = True,
task_id = "fail",
python_callable = all_bad
)

passs = PythonOperator(
provide_context = True,
task_id = "pass",
python_callable = all_good
)

start >> [passs, fail]

From the dag processor logs:

The context that is passed is from the task instance that has succeeded.

[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: <function failure_callback_dag at 0x7fd17ca18560>
[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 - <TaskInstance: test_dag.pass manual__2022-09-28T18:27:59.612118+00:00 [success]>
[2022-09-28T18:28:14.944+0000] {logging_mixin.py:117} INFO - <Task(PythonOperator): pass>

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

Default providers that are present in the official airflow docker image.

Deployment

Docker-Compose

Deployment details

No response

Anything else

Not sure if it is an expected behaviour, incase it is it needs to be documented in Callbacks.

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