<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">The <code class="notranslate">on_failure_callback</code> 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.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The <code class="notranslate">on_failure_callback</code> should only be invoked once</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Here is the DAG which triggers the effect</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import pendulum
from datetime import timedelta
from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator
import logging
import requests

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

DatabricksSubmitRunOperator(
    task_id=&quot;example_task&quot;,
    dag=dag,
    databricks_conn_id=&quot;databricks_default&quot;,
    spark_submit_task={
        &quot;parameters&quot;:  [
            &quot;--class&quot;, &quot;com.example.Launcher&quot;,
            f&quot;dbfs:/libraries/scala/example-fat-jar.jar&quot;
        ]
    },
    new_cluster={
        &quot;spark_version&quot;:
        &quot;10.4.x-cpu-ml-scala2.12&quot;,
        &quot;spark_env_vars&quot;: {
            &quot;JNAME&quot;: &quot;zulu11-ca-amd64&quot;  # Use JDK 11
        },
        &quot;spark_conf&quot;: {
            &quot;spark.sql.session.timeZone&quot;: &quot;UTC&quot;
        },
        &quot;aws_attributes&quot;: {
            &quot;instance_profile_arn&quot;: &quot;arn:aws:iam::765&lt;removed-for-publication&gt;:instance-profile/DatabricksExecution&quot;
        },
        &quot;instance_pool_id&quot;: &quot;1011-&lt;removed-for-publication&gt;&quot;,
        &quot;driver_instance_pool_id&quot;: &quot;1011-&lt;removed-for-publication&gt;&quot;,
        &quot;num_workers&quot;: 1
    }
)"><pre class="notranslate"><code class="notranslate">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://&lt;removed-for-publication&gt;.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&lt;removed-for-publication&gt;:instance-profile/DatabricksExecution"
        },
        "instance_pool_id": "1011-&lt;removed-for-publication&gt;",
        "driver_instance_pool_id": "1011-&lt;removed-for-publication&gt;",
        "num_workers": 1
    }
)
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">composer-2.0.32</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">Here is the full <code class="notranslate">requirements.txt</code></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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"><pre class="notranslate"><code class="notranslate">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
</code></pre></div>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Composer</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">We are running a Cloud Compose environment with image <code class="notranslate">composer-2.0.32-airflow-2.3.4</code></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Since logging inside the callback is not working for us (See <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1578233527" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/29442" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/29442/hovercard" href="https://github.com/apache/airflow/issues/29442">#29442</a>), here is a screenshot from RequestBin. Both invocations share the same log line, the context and stack trace is also the same.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/3661031/218069601-dada3f0e-4494-4cf3-a274-f5d3b8270ac8.png"><img src="https://user-images.githubusercontent.com/3661031/218069601-dada3f0e-4494-4cf3-a274-f5d3b8270ac8.png" alt="image" style="max-width: 100%;"></a></p>
<p dir="auto">Here is the complete stack trace for either invocation.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="File &quot;/opt/python3.8/bin/airflow&quot;, line 8, in &lt;module&gt;
    sys.exit(main())
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/__main__.py&quot;, line 38, in main
    args.func(args)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/cli/cli_parser.py&quot;, line 51, in command
    return func(*args, **kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/utils/cli.py&quot;, line 101, in wrapper
    return f(*args, **kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 76, in scheduler
    _run_scheduler_job(args=args)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/cli/commands/scheduler_command.py&quot;, line 46, in _run_scheduler_job
    job.run()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/jobs/base_job.py&quot;, line 244, in run
    self._execute()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py&quot;, line 748, in _execute
    self.processor_agent.start()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py&quot;, line 158, in start
    process.start()
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/process.py&quot;, line 121, in start
    self._popen = self._Popen(self)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/context.py&quot;, line 277, in _Popen
    return Popen(process_obj)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py&quot;, line 19, in __init__
    self._launch(process_obj)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py&quot;, line 75, in _launch
    code = process_obj._bootstrap(parent_sentinel=child_r)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py&quot;, line 255, in _run_processor_manager
    processor_manager.start()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py&quot;, line 486, in start
    return self._run_parsing_loop()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py&quot;, line 600, in _run_parsing_loop
    self.start_new_processes()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/manager.py&quot;, line 1003, in start_new_processes
    processor.start()
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py&quot;, line 194, in start
    process.start()
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/process.py&quot;, line 121, in start
    self._popen = self._Popen(self)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/context.py&quot;, line 277, in _Popen
    return Popen(process_obj)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py&quot;, line 19, in __init__
    self._launch(process_obj)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/popen_fork.py&quot;, line 75, in _launch
    code = process_obj._bootstrap(parent_sentinel=child_r)
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/python3.8/lib/python3.8/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py&quot;, line 155, in _run_file_processor
    result: Tuple[int, int] = dag_file_processor.process_file(
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py&quot;, line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py&quot;, line 656, in process_file
    self.execute_callbacks(dagbag, callback_requests)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py&quot;, line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py&quot;, line 581, in execute_callbacks
    self._execute_dag_callbacks(dagbag, request, session)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/dag_processing/processor.py&quot;, line 595, in _execute_dag_callbacks
    dag.handle_callback(
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/python3.8/lib/python3.8/site-packages/airflow/models/dag.py&quot;, line 1178, in handle_callback
    callback(context)
  File &quot;/home/airflow/gcs/dags/integration_test/example.py&quot;"><pre class="notranslate"><code class="notranslate">File "/opt/python3.8/bin/airflow", line 8, in &lt;module&gt;
    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"
</code></pre></div>
<p dir="auto">Here is the log for the Databricks task that was marked skipped, because the DAG as a whole timed out.</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="*** Reading remote log from gs://europe-west3-airflow-&lt;removed-for-publication&gt;-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 &lt;TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [queued]&gt;
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1172} INFO - Dependencies all met for &lt;TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [queued]&gt;
[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 &lt;Task(DatabricksSubmitRunOperator): example_task&gt; 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 &lt;TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [running]&gt; 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 &lt;removed-for-publication&gt;
[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 &lt;removed-for-publication&gt;
[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 &lt;removed-for-publication&gt;
[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"><pre class="notranslate"><code class="notranslate">*** Reading remote log from gs://europe-west3-airflow-&lt;removed-for-publication&gt;-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 &lt;TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [queued]&gt;
[2023-02-10, 09:34:42 UTC] {taskinstance.py:1172} INFO - Dependencies all met for &lt;TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [queued]&gt;
[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 &lt;Task(DatabricksSubmitRunOperator): example_task&gt; 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 &lt;TaskInstance: INT_minimal_example.example_task manual__2023-02-10T09:34:38+00:00 [running]&gt; 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 &lt;removed-for-publication&gt;
[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 &lt;removed-for-publication&gt;
[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 &lt;removed-for-publication&gt;
[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
</code></pre></div>
<p dir="auto">Finally, a screenshot of the DAG failure for timing purposes:</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/3661031/218072255-4ab3f608-8dfc-442a-b59f-c87c4b20f834.png"><img src="https://user-images.githubusercontent.com/3661031/218072255-4ab3f608-8dfc-442a-b59f-c87c4b20f834.png" alt="image" style="max-width: 100%;"></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"> 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.2</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I have applied task callback for failure and success case using Cluster policy for the specific dag <a href="https://gist.github.com/Bowrna/7fdca8b546fc274edd068ffdae5b76f9">https://gist.github.com/Bowrna/7fdca8b546fc274edd068ffdae5b76f9</a></p>
<p dir="auto">I am attaching the cluster policy that I have applied here.<br>
<a href="https://gist.github.com/Bowrna/1994894beea39fa8e1c269591b7f0346">https://gist.github.com/Bowrna/1994894beea39fa8e1c269591b7f0346</a></p>
<p dir="auto">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</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Like success callback, failure callback also should get executed only once.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">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</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062485, &quot;task_name&quot;: &quot;runme_0&quot;, &quot;task_duration&quot;: 1.178738, &quot;task_status&quot;: &quot;success&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062486, &quot;task_name&quot;: &quot;runme_1&quot;, &quot;task_duration&quot;: 1.081635, &quot;task_status&quot;: &quot;success&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062488, &quot;task_name&quot;: &quot;runme_2&quot;, &quot;task_duration&quot;: 1.071424, &quot;task_status&quot;: &quot;success&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062495, &quot;task_name&quot;: &quot;also_run_this&quot;, &quot;task_duration&quot;: 0.075827, &quot;task_status&quot;: &quot;success&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062495, &quot;task_name&quot;: &quot;this_will_skip&quot;, &quot;task_duration&quot;: 0.072133, &quot;task_status&quot;: &quot;skipped&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;failure&quot;, &quot;unravel_timestamp&quot;: 1668062496, &quot;task_name&quot;: &quot;failure_case&quot;, &quot;task_duration&quot;: 0.066066, &quot;task_status&quot;: &quot;failed&quot;, &quot;task_operator&quot;: &quot;PythonOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;failure&quot;, &quot;unravel_timestamp&quot;: 1668062497, &quot;task_name&quot;: &quot;failure_case&quot;, &quot;task_duration&quot;: 0.655575, &quot;task_status&quot;: &quot;failed&quot;, &quot;task_operator&quot;: &quot;PythonOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062498, &quot;task_name&quot;: &quot;run_after_loop&quot;, &quot;task_duration&quot;: 0.073114, &quot;task_status&quot;: &quot;success&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}
{&quot;dag_name&quot;: &quot;test_bowrna&quot;, &quot;dag_run_name&quot;: &quot;scheduled__2022-11-09T00:00:00+00:00&quot;, &quot;status_callback&quot;: &quot;success&quot;, &quot;unravel_timestamp&quot;: 1668062500, &quot;task_name&quot;: &quot;run_this_last&quot;, &quot;task_duration&quot;: 0.078573, &quot;task_status&quot;: &quot;success&quot;, &quot;task_operator&quot;: &quot;BashOperator&quot;, &quot;dag_start_date&quot;: &quot;2022-11-10 06:41:22.757324+00:00&quot;, &quot;dag_end_date&quot;: null, &quot;dag_state&quot;: &quot;running&quot;}"><pre class="notranslate"><code class="notranslate">{"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"}
</code></pre></div>
<p dir="auto">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)</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">macOS Monterey version 12.2.1</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-apache-hive==4.0.1<br>
apache-airflow-providers-cncf-kubernetes==4.4.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-sqlite==3.2.1</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Virtualenv installation</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">I am currently running the airflow in development mode and testing callback and cluster policy</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Once it is getting executed as part of the <code class="notranslate">taskinstance</code> callback, while another time it is getting executed as part of <code class="notranslate">DagFileProcessorProcess</code><br>
I pulled out the logs from the specific task that is getting logged twice and this is part of <code class="notranslate">taskinstance.py</code> flow</p>
<p dir="auto">I found the below log in the task of the specific dag_run</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/Users/unravel/unravel_airflow/unravel_airflow/lib/python3.10/site-packages/airflow/operators/python.py&quot;, line 175, in execute
    return_value = self.execute_callable()
  File &quot;/Users/unravel/unravel_airflow/unravel_airflow/lib/python3.10/site-packages/airflow/operators/python.py&quot;, line 193, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File &quot;/Users/unravel/unravel_airflow/airflow/dags/test.py&quot;, 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 &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [running]&gt; 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: &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt;
[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: &lt;function task_failure_alert at 0x10749a050&gt;
[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': &lt;TaskInstanceState.FAILED: 'failed'&gt;, '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 [&lt;TaskInstanceState.FAILED: 'failed'&gt;] unchanged () deleted ['running'] - &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt;
[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 &lt;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&gt;: 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"><pre lang="[2022-11-09," class="notranslate"><code class="notranslate">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 &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [running]&gt; 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: &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt;
[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: &lt;function task_failure_alert at 0x10749a050&gt;
[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': &lt;TaskInstanceState.FAILED: 'failed'&gt;, '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 [&lt;TaskInstanceState.FAILED: 'failed'&gt;] unchanged () deleted ['running'] - &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt;
[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 &lt;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&gt;: 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
</code></pre></div>
<p dir="auto">Also I found another log in <code class="notranslate">airflow scheduler</code></p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[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 &lt;DAG: test_bowrna&gt;
[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': &quot;{'DAG Id': 'test_bowrna', 'Task Id': 'failure_case', 'Run Id': 'manual__2022-11-09T09:59:32.540415+00:00', 'Hostname': 'unravels-macbook-pro.local'}&quot;, 'simple_task_instance': &lt;airflow.models.taskinstance.SimpleTaskInstance object at 0x1049f7a60&gt;, '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 &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [running]&gt; 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: &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt;
[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: &lt;function task_failure_alert at 0x1048c9240&gt;
[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': &lt;TaskInstanceState.FAILED: 'failed'&gt;, '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 &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt; 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"><pre class="notranslate"><code class="notranslate">[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 &lt;DAG: test_bowrna&gt;
[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': &lt;airflow.models.taskinstance.SimpleTaskInstance object at 0x1049f7a60&gt;, '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 &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [running]&gt; 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: &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt;
[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: &lt;function task_failure_alert at 0x1048c9240&gt;
[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': &lt;TaskInstanceState.FAILED: 'failed'&gt;, '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 &lt;TaskInstance: test_bowrna.failure_case manual__2022-11-09T09:59:32.540415+00:00 [failed]&gt; 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
</code></pre></div>
<p dir="auto">This also logs the failure case that gets executed from processor.py flow of the code</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>