callback functions not called when a dag run is marked success or failure

[lssatvik]

Apache Airflow version

2.1.1

Operating System

Ubuntu 20.04.3 LTS

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.0.0
apache-airflow-providers-celery==2.0.0
apache-airflow-providers-cncf-kubernetes==2.0.0
apache-airflow-providers-databricks==2.0.0
apache-airflow-providers-ftp==2.0.0
apache-airflow-providers-imap==2.0.0
apache-airflow-providers-mysql==2.0.0
apache-airflow-providers-postgres==2.0.0
apache-airflow-providers-redis==2.0.0
apache-airflow-providers-sqlite==2.0.0
apache-airflow-providers-ssh==2.0.0

Deployment

Virtualenv installation

Deployment details

No response

What happened

'on_success_callback' and 'on_failure_callback' functions were not called when manually marking success or failure on a completed dag run.

What you expected to happen

'on_success_callback' and 'on_failure_callback' functions should be called whenever marking success or failure on a dag run even after it completes.

How to reproduce

Provide callback function as 'on_success_callback' argument when initializing DAG operator. Let one dag run complete. Check if 'on_success_callback' function is called. Then mark the dag run as failed. Then mark the dag run as succeeded. The 'on_success_callback' function is not called.

Anything else

The callback functions are called when a dagrun is automatically marked success/failure by scheduler but aren't called when manually setting them after run completion. I have read earlier posts on stack overflow that this was not intended behaviour in Airflow 1 but could not find similar posts for Airflow 2 if this is a bug and how to fix it.

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[eladkal]

I think this is by design. Marking the DAG Run as success/failure just changes the record status in the database.
Similarly - marking task as failed does not invoke the operator on_kill()

[eladkal]

The question if it's a bug or a feature request is relevant also for the API.
in Airflow 2.2 we added endpoint #17839 that also just change the dag run status.
@ephraimbuddy @uranusjr can you share your thoughts on this?

[potiuk]

The problem with running the callbacks on UI or API action is that those have different "environment"/context than the regular callbacks.

When the task fails or succeeds during the normal execution, the callbacks are run in workers or Kubernetes PODs or Local Executor processes. When you change the state via UI or API, the callback (if we naively implement it) will be run in webserver. This is not good (and is a security issue if we allow for that). No DAG-provided code should ever be run in the webserver.

Implementing callback "properly" in this case would require to actually schedule another "task" execution in the worker an run it there. Which is expensive and rather complex operation to do. So I see that as a current "feature" rather than "bug".

However things might change possibly with Airflow 2.1.1 and the Triggerer / Defferable Operator approach possibly. I believer (@andrewgodwin ?) it would be much easier to use the mechanisms we have added for that AIP for that purpose.

[andrewgodwin]

Unfortunately the triggerer mechanism doesn't solve this - triggers can run separately, but even to raise a task failure, deferred tasks have to be sent back to a worker to execute, so they're still bound by the same sort of rules.

One thing I do want to look at soon, however, is extending the Executor interface to allow executors to run more than just Tasks - in essence, opening up the ability to submit other workloads (like callback functions) to run on executors, and hopefully pave the way for moving these and a lot of other things out of the scheduler/webserver.

[potiuk]

One thing I do want to look at soon, however, is extending the Executor interface to allow executors to run more than just Tasks - in essence, opening up the ability to submit other workloads (like callback functions) to run on executors, and hopefully pave the way for moving these and a lot of other things out of the scheduler/webserver.

Sounds great,

[lssatvik]

I am upgrading to Airflow 2.2.2, and I observed that the success callback is being executed when marking success while dag is running. It is not executed in 2.1.1. So the feature has been implemented then? If yes, I shall close this request.

[potiuk]

Let;s close this. I think it should have worked already. It's been added here: #10917 - apparently in 2.0.1 already . I am not sure why it has not been executed in 2.1.1 but apparently id works as expected. We can close that one. and reopen if we find it is a problem in some cases (but need reproduction scenario).