Issue 1:
Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

The on_failure_callback is invoked twice when a DAG fails due to a timeout. This leads to duplicate failure alerts in our Slack channels. It happens reliably and the invocations are roughly 5 seconds apart. Context and full stack trace in both invocations are identical.

What you think should happen instead

The on_failure_callback should only be invoked once

How to reproduce

Here is the DAG which triggers the effect

import pendulum
from datetime import timedelta
from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator
import logging
import requests

dag_settings = {
"dag_id": f"INT_minimal_example",
"max_active_runs": 1,
"dagrun_timeout": timedelta(minutes=1),
"start_date": pendulum.today("UTC"),
"default_args": {
"owner": "airflow",
"catchup": False
},
"tags": ["env:testing"],
"on_failure_callback": (
lambda context: [
logging.info(f"Minimal example - failure callback with context: {context}"),
# Since logging.info does not work for us, we add a request to RequestBin
# See https://github.com/apache/airflow/issues/29442
requests.post(
"https://<removed-for-publication>.m.pipedream.net",
json={"payload": f"Minimal example - failure callback with context: {context}"}
)
]
)
}
dag = DAG(**dag_settings)
logging.info(f"Minimal example - Created DAG {dag.dag_id}.")

DatabricksSubmitRunOperator(
task_id="example_task",
dag=dag,
databricks_conn_id="databricks_default",
spark_submit_task={
"parameters": [
"--class", "com.example.Launcher",
f"dbfs:/libraries/scala/example-fat-jar.jar"
]
},
new_cluster={
"spark_version":
"10.4.x-cpu-ml-scala2.12",
"spark_env_vars": {
"JNAME": "zulu11-ca-amd64" # Use JDK 11
},
"spark_conf": {
"spark.sql.session.timeZone": "UTC"
},
"aws_attributes": {
"instance_profile_arn": "arn:aws:iam::765<removed-for-publication>:instance-profile/DatabricksExecution"
},
"instance_pool_id": "1011-<removed-for-publication>",
"driver_instance_pool_id": "1011-<removed-for-publication>",
"num_workers": 1
}
)

Operating System

composer-2.0.32

Versions of Apache Airflow Providers

Here is the full requirements.txt

apache-airflow-providers-databricks==4.0.0
databricks-sql-connector==2.1.0
apache-beam~=2.43.0
sqlalchemy-bigquery==1.5.0
requests~=2.28.1
apache-airflow-providers-tableau==4.0.0
apache-airflow-providers-sendgrid==3.1.0
python-dotenv==0.21.0
urllib3~=1.26.8
tableauserverclient==0.23
apache-airflow-providers-http==4.1.0
# time library in airflow
pendulum==2.1.2

Deployment

Composer

Deployment details

We are running a Cloud Compose environment with image composer-2.0.32-airflow-2.3.4

Anything else

Since logging inside the callback is not working for us (See #29442), here is a screenshot from RequestBin. Both invocations share the same log line, the context and stack trace is also the same.

Here is the complete stack trace for either invocation.

File "/opt/python3.8/bin/airflow", line 8, in <module>
sys.exit(main())
File "/opt/python3.8/lib/python3.8/site-packages/airflow/__main__.py", line 38, in main
args.func(args)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 51, in command
return func(*args, **kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/cli.py", line 101, in wrapper
return f(*args, **kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/cli/commands/scheduler_command.py", line 76, in scheduler
_run_scheduler_job(args=args)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/cli/commands/scheduler_command.py", line 46, in _run_scheduler_job
job.run()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/jobs/base_job.py", line 244, in run
self._execute()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 748, in _execute
self.processor_agent.start()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 158, in start
process.start()
File "/opt/python3.8/lib/python3.8/multiprocessing/process.py", line 121, in start
self._popen = self._Popen(self)
File "/opt/python3.8/lib/python3.8/multiprocessing/context.py", line 277, in _Popen
return Popen(process_obj)
File "/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py", line 19, in __init__
self._launch(process_obj)
File "/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py", line 75, in _launch
code = process_obj._bootstrap(parent_sentinel=child_r)
File "/opt/python3.8/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
self.run()
File "/opt/python3.8/lib/python3.8/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 255, in _run_processor_manager
processor_manager.start()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 486, in start
return self._run_parsing_loop()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 600, in _run_parsing_loop
self.start_new_processes()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py", line 1003, in start_new_processes
processor.start()
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 194, in start
process.start()
File "/opt/python3.8/lib/python3.8/multiprocessing/process.py", line 121, in start
self._popen = self._Popen(self)
File "/opt/python3.8/lib/python3.8/multiprocessing/context.py", line 277, in _Popen
return Popen(process_obj)
File "/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py", line 19, in __init__
self._launch(process_obj)
File "/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py", line 75, in _launch
code = process_obj._bootstrap(parent_sentinel=child_r)
File "/opt/python3.8/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
self.run()
File "/opt/python3.8/lib/python3.8/multiprocessing/process.py", line 108, in run
self._target(*self._args, **self._kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 155, in _run_file_processor
result: Tuple[int, int] = dag_file_processor.process_file(
File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py", line 71, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 656, in process_file
self.execute_callbacks(dagbag, callback_requests)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py", line 71, in wrapper
return func(*args, session=session, **kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 581, in execute_callbacks
self._execute_dag_callbacks(dagbag, request, session)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py", line 68, in wrapper
return func(*args, **kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py", line 595, in _execute_dag_callbacks
dag.handle_callback(
File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py", line 68, in wrapper
return func(*args, **kwargs)
File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/dag.py", line 1178, in handle_callback
callback(context)
File "/home/airflow/gcs/dags/integration_test/example.py"

Here is the log for the Databricks task that was marked skipped, because the DAG as a whole timed out.

*** Reading remote log from gs://europe-west3-airflow-<removed-for-publication>-bucket/logs/dag_id=INT_minimal_example/run_id=manual__2023-02-10T09:34:38+00:00/task_id=example_task/attempt=1.log.
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1172} INFO - Dependencies all met for <TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [queued]>
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1172} INFO - Dependencies all met for <TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [queued]>
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1369} INFO -
--------------------------------------------------------------------------------
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1370} INFO - Starting attempt 1 of 3
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1371} INFO -
--------------------------------------------------------------------------------
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1390} INFO - Executing <Task(DatabricksSubmitRunOperator): example_task> on 2023-02-10 09:34:38+00:00
[2023-02-10, 09:34:42 UTC] {standard_task_runner.py:52} INFO - Started process 69373 to run task
[2023-02-10, 09:34:42 UTC] {standard_task_runner.py:79} INFO - Running: ['airflow', 'tasks', 'run', 'INT_minimal_example', 'example_task', 'manual__2023-02-10T09:34:38+00:00', '--job-id', '197269', '--raw', '--subdir', 'DAGS_FOLDER/integration_test/minimal_example.py', '--cfg-path', '/tmp/tmp4ffeji6u', '--error-file', '/tmp/tmpd5pieigz']
[2023-02-10, 09:34:42 UTC] {standard_task_runner.py:80} INFO - Job 197269: Subtask example_task
[2023-02-10, 09:34:43 UTC] {task_command.py:375} INFO - Running <TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [running]> on host airflow-worker-lhbx8
[2023-02-10, 09:34:43 UTC] {taskinstance.py:1583} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=INT_minimal_example
AIRFLOW_CTX_TASK_ID=example_task
AIRFLOW_CTX_EXECUTION_DATE=2023-02-10T09:34:38+00:00
AIRFLOW_CTX_TRY_NUMBER=1
AIRFLOW_CTX_DAG_RUN_ID=manual__2023-02-10T09:34:38+00:00
[2023-02-10, 09:34:43 UTC] {base.py:68} INFO - Using connection ID 'databricks_default' for task execution.
[2023-02-10, 09:34:43 UTC] {databricks_base.py:333} INFO - Using basic auth.
[2023-02-10, 09:34:43 UTC] {databricks.py:75} INFO - Run submitted with run_id: 102832491
[2023-02-10, 09:34:43 UTC] {databricks_base.py:333} INFO - Using basic auth.
[2023-02-10, 09:34:43 UTC] {databricks_base.py:333} INFO - Using basic auth.
[2023-02-10, 09:34:43 UTC] {databricks.py:97} INFO - example_task in run state: {'life_cycle_state': 'PENDING', 'result_state': '', 'state_message': ''}
[2023-02-10, 09:34:43 UTC] {databricks.py:98} INFO - View run status, Spark UI, and logs at <removed-for-publication>
[2023-02-10, 09:34:43 UTC] {databricks.py:99} INFO - Sleeping for 30 seconds.
[2023-02-10, 09:35:13 UTC] {databricks_base.py:333} INFO - Using basic auth.
[2023-02-10, 09:35:13 UTC] {databricks.py:97} INFO - example_task in run state: {'life_cycle_state': 'PENDING', 'result_state': '', 'state_message': 'Waiting for cluster'}
[2023-02-10, 09:35:13 UTC] {databricks.py:98} INFO - View run status, Spark UI, and logs at <removed-for-publication>
[2023-02-10, 09:35:13 UTC] {databricks.py:99} INFO - Sleeping for 30 seconds.
[2023-02-10, 09:35:43 UTC] {databricks_base.py:333} INFO - Using basic auth.
[2023-02-10, 09:35:44 UTC] {databricks.py:97} INFO - example_task in run state: {'life_cycle_state': 'PENDING', 'result_state': '', 'state_message': 'Waiting for cluster'}
[2023-02-10, 09:35:44 UTC] {databricks.py:98} INFO - View run status, Spark UI, and logs at <removed-for-publication>
[2023-02-10, 09:35:44 UTC] {databricks.py:99} INFO - Sleeping for 30 seconds.
[2023-02-10, 09:35:44 UTC] {local_task_job.py:225} WARNING - DagRun timed out after 0:01:05.297413.
[2023-02-10, 09:35:44 UTC] {local_task_job.py:226} WARNING - State of this instance has been externally set to skipped. Terminating instance.
[2023-02-10, 09:35:44 UTC] {process_utils.py:125} INFO - Sending Signals.SIGTERM to group 69373. PIDs of all processes in the group: [69373]
[2023-02-10, 09:35:44 UTC] {process_utils.py:80} INFO - Sending the signal Signals.SIGTERM to group 69373
[2023-02-10, 09:35:44 UTC] {taskinstance.py:1555} ERROR - Received SIGTERM. Terminating subprocesses.
[2023-02-10, 09:35:44 UTC] {base.py:68} INFO - Using connection ID 'databricks_default' for task execution.
[2023-02-10, 09:35:44 UTC] {databricks_base.py:333} INFO - Using basic auth.
[2023-02-10, 09:35:44 UTC] {databricks.py:345} INFO - Task: example_task with run_id: 102832491 was requested to be cancelled.
[2023-02-10, 09:35:44 UTC] {process_utils.py:75} INFO - Process psutil.Process(pid=69373, status='terminated', exitcode=0, started='09:34:41') (69373) terminated with exit code 0

Finally, a screenshot of the DAG failure for timing purposes:

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

What happened

I have applied task callback for failure and success case using Cluster policy for the specific dag https://gist.github.com/Bowrna/7fdca8b546fc274edd068ffdae5b76f9

I am attaching the cluster policy that I have applied here.

https://gist.github.com/Bowrna/1994894beea39fa8e1c269591b7f0346

On executing the DAG, the success callback is correctly invoked once for every successful task, while failure callback is invoked twice for a failure in task

What you think should happen instead

Like success callback, failure callback also should get executed only once.

How to reproduce

I have attached the sample DAG and airflow_local_settings.py file in which i have added the cluster policy used. On running airflow with that and executing the DAG either manually/scheduled will cause to log the below details

{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062485, "task_name": "runme_0", "task_duration": 1.178738, "task_status": "success", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062486, "task_name": "runme_1", "task_duration": 1.081635, "task_status": "success", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062488, "task_name": "runme_2", "task_duration": 1.071424, "task_status": "success", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062495, "task_name": "also_run_this", "task_duration": 0.075827, "task_status": "success", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062495, "task_name": "this_will_skip", "task_duration": 0.072133, "task_status": "skipped", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "failure", "unravel_timestamp": 1668062496, "task_name": "failure_case", "task_duration": 0.066066, "task_status": "failed", "task_operator": "PythonOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "failure", "unravel_timestamp": 1668062497, "task_name": "failure_case", "task_duration": 0.655575, "task_status": "failed", "task_operator": "PythonOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062498, "task_name": "run_after_loop", "task_duration": 0.073114, "task_status": "success", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}
{"dag_name": "test_bowrna", "dag_run_name": "scheduled__2022-11-09T00:00:00+00:00", "status_callback": "success", "unravel_timestamp": 1668062500, "task_name": "run_this_last", "task_duration": 0.078573, "task_status": "success", "task_operator": "BashOperator", "dag_start_date": "2022-11-10 06:41:22.757324+00:00", "dag_end_date": null, "dag_state": "running"}

Here you can see for the failure case, it is logged twice. Also both of these failure case logging have different duration(I have verified if i am double logging in code by mistake, but that is not the case)

Operating System

macOS Monterey version 12.2.1

Versions of Apache Airflow Providers

apache-airflow-providers-apache-hive==4.0.1

apache-airflow-providers-cncf-kubernetes==4.4.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-sqlite==3.2.1

Deployment

Virtualenv installation

Deployment details

I am currently running the airflow in development mode and testing callback and cluster policy

Anything else

Once it is getting executed as part of the taskinstance callback, while another time it is getting executed as part of DagFileProcessorProcess

I pulled out the logs from the specific task that is getting logged twice and this is part of taskinstance.py flow

I found the below log in the task of the specific dag_run

Traceback (most recent call last):
File "/Users/unravel/unravel_airflow/unravel_airflow/lib/python3.10/site-packages/airflow/operators/python.py", line 175, in execute
return_value = self.execute_callable()
File "/Users/unravel/unravel_airflow/unravel_airflow/lib/python3.10/site-packages/airflow/operators/python.py", line 193, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File "/Users/unravel/unravel_airflow/airflow/dags/test.py", line 67, in fail_case
raise AirflowFailException('Failure case test for cluster policy')
airflow.exceptions.AirflowFailException: Failure case test for cluster policy
[2022-11-09, 09:59:40 UTC] {taskinstance.py:851} DEBUG - Refreshing TaskInstance <TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [running]> from DB
[2022-11-09, 09:59:40 UTC] {taskinstance.py:2325} DEBUG - Task Duration set to 0.061723
[2022-11-09, 09:59:40 UTC] {taskinstance.py:1412} DEBUG - Clearing next_method and next_kwargs.
[2022-11-09, 09:59:40 UTC] {taskinstance.py:1401} INFO - Immediate failure requested. Marking task as FAILED. dag_id=test_bowrna, task_id=failure_case, execution_date=20221109T095932, start_date=20221109T095940, end_date=20221109T095940
[2022-11-09, 09:59:40 UTC] {logging_mixin.py:120} INFO - Cluster dag policy:Task has failed
[2022-11-09, 09:59:40 UTC] {logging_mixin.py:120} INFO - Cluster policy:context for failure case: <TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]>
[2022-11-09, 09:59:40 UTC] {logging_mixin.py:120} INFO - Task id: failure_case
[2022-11-09, 09:59:40 UTC] {logging_mixin.py:120} INFO - Task id: <function task_failure_alert at 0x10749a050>
[2022-11-09, 09:59:40 UTC] {logging_mixin.py:120} INFO - Path to write: /Users/unravel/unravel_airflow/airflow/event_logger/test_bowrna/09-11-2022/manual__2022-11-09T09:59:32.540415+00:00.json
[2022-11-09, 09:59:40 UTC] {logging_mixin.py:120} INFO - Info details to log: {'dag_name': 'test_bowrna', 'dag_run_name': 'manual__2022-11-09T09:59:32.540415+00:00', 'status_callback': 'failure', 'unravel_timestamp': 1667987980, 'task_name': 'failure_case', 'task_duration': 0.061723, 'task_status': <TaskInstanceState.FAILED: 'failed'>, 'task_operator': 'PythonOperator', 'dag_start_date': datetime.datetime(2022, 11, 9, 9, 59, 34, 24988, tzinfo=Timezone('UTC')), 'dag_end_date': None, 'dag_state': 'running'}
[2022-11-09, 09:59:40 UTC] {events.py:45} DEBUG - session flush listener: added [<TaskInstanceState.FAILED: 'failed'>] unchanged () deleted ['running'] - <TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]>
[2022-11-09, 09:59:40 UTC] {cli_action_loggers.py:83} DEBUG - Calling callbacks: []
[2022-11-09, 09:59:40 UTC] {standard_task_runner.py:100} ERROR - Failed to execute job 1318 for task failure_case ('TaskInstance' object has no attribute 'task'; 49534)
[2022-11-09, 09:59:40 UTC] {local_task_job.py:164} INFO - Task exited with return code 1
[2022-11-09, 09:59:40 UTC] {dagrun.py:674} DEBUG - number of tis tasks for <DagRun test_bowrna @ 2022-11-09 09:59:32.540415+00:00: manual__2022-11-09T09:59:32.540415+00:00, state:running, queued_at: 2022-11-09 09:59:32.551538+00:00. externally triggered: True>: 0 task(s)
[2022-11-09, 09:59:40 UTC] {local_task_job.py:273} INFO - 0 downstream tasks scheduled from follow-on schedule check

Also I found another log in airflow scheduler

[2022-11-09T15:29:41.148+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.148+0530] {[dagbag.py:502](http://dagbag.py:502/)} DEBUG - Loaded DAG <DAG: test_bowrna>
[2022-11-09T15:29:41.149+0530] {[processor.py:766](http://processor.py:766/)} INFO - DAG(s) dict_keys(['test_bowrna']) retrieved from /Users/unravel/unravel_airflow/airflow/dags/test.py
[2022-11-09T15:29:41.149+0530] {[processor.py:639](http://processor.py:639/)} DEBUG - Processing Callback Request: {'full_filepath': '/Users/unravel/unravel_airflow/airflow/dags/test.py', 'processor_subdir': '/Users/unravel/unravel_airflow/airflow/dags', 'msg': "{'DAG Id': 'test_bowrna', 'Task Id': 'failure_case', 'Run Id': 'manual__2022-11-09T09:59:32.540415+00:00', 'Hostname': 'unravels-macbook-pro.local'}", 'simple_task_instance': <airflow.models.taskinstance.SimpleTaskInstance object at 0x1049f7a60>, 'is_failure_callback': True}
[2022-11-09T15:29:41.198+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.198+0530] {[taskinstance.py:1853](http://taskinstance.py:1853/)} ERROR - {'DAG Id': 'test_bowrna', 'Task Id': 'failure_case', 'Run Id': 'manual__2022-11-09T09:59:32.540415+00:00', 'Hostname': 'unravels-macbook-pro.local'}
[2022-11-09T15:29:41.199+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.199+0530] {[taskinstance.py:851](http://taskinstance.py:851/)} DEBUG - Refreshing TaskInstance <TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [running]> from DB
[2022-11-09T15:29:41.201+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.201+0530] {[taskinstance.py:2325](http://taskinstance.py:2325/)} DEBUG - Task Duration set to 0.520437
[2022-11-09T15:29:41.201+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.201+0530] {[taskinstance.py:1412](http://taskinstance.py:1412/)} DEBUG - Clearing next_method and next_kwargs.
[2022-11-09T15:29:41.201+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.201+0530] {plugins_manager.py:300} DEBUG - Loading plugins
[2022-11-09T15:29:41.201+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.201+0530] {plugins_manager.py:244} DEBUG - Loading plugins from directory: /Users/unravel/unravel_airflow/airflow/plugins
[2022-11-09T15:29:41.202+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.202+0530] {plugins_manager.py:259} DEBUG - Importing plugin module /Users/unravel/unravel_airflow/airflow/plugins/plugin.py
[2022-11-09T15:29:41.202+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.202+0530] {plugins_manager.py:259} DEBUG - Importing plugin module /Users/unravel/unravel_airflow/airflow/plugins/listener.py
[2022-11-09T15:29:41.204+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.203+0530] {plugins_manager.py:224} DEBUG - Loading plugins from entrypoints
[2022-11-09T15:29:41.228+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.228+0530] {plugins_manager.py:316} DEBUG - Loading 1 plugin(s) took 0.03 seconds
[2022-11-09T15:29:41.228+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.228+0530] {plugins_manager.py:476} DEBUG - Integrate DAG plugins
[2022-11-09T15:29:41.230+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.230+0530] {[taskinstance.py:1099](http://taskinstance.py:1099/)} DEBUG - previous_execution_date was called
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.231+0530] {[taskinstance.py:1401](http://taskinstance.py:1401/)} INFO - Marking task as FAILED. dag_id=test_bowrna, task_id=failure_case, execution_date=20221109T095932, start_date=20221109T095940, end_date=20221109T095941
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - Cluster dag policy:Task has failed
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - Cluster policy:context for failure case: <TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]>
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - Task id: failure_case
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - Task id: <function task_failure_alert at 0x1048c9240>
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - Path to write: /Users/unravel/unravel_airflow/airflow/event_logger/test_bowrna/09-11-2022/manual__2022-11-09T09:59:32.540415+00:00.json
[2022-11-09T15:29:41.231+0530] {logging_mixin.py:120} INFO - Info details to log: {'dag_name': 'test_bowrna', 'dag_run_name': 'manual__2022-11-09T09:59:32.540415+00:00', 'status_callback': 'failure', 'unravel_timestamp': 1667987981, 'task_name': 'failure_case', 'task_duration': 0.520437, 'task_status': <TaskInstanceState.FAILED: 'failed'>, 'task_operator': 'PythonOperator', 'dag_start_date': datetime.datetime(2022, 11, 9, 9, 59, 34, 24988, tzinfo=Timezone('UTC')), 'dag_end_date': None, 'dag_state': 'running'}
[2022-11-09T15:29:41.234+0530] {[processor.py:725](http://processor.py:725/)} INFO - Executed failure callback for <TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]> in state failed
[2022-11-09T15:29:41.236+0530] {logging_mixin.py:120} INFO - [2022-11-09T15:29:41.236+0530] {[dagbag.py:647](http://dagbag.py:647/)} DEBUG - Running dagbag.sync_to_db with retries. Try 1 of

This also logs the failure case that gets executed from processor.py flow of the code

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