Introduce generic Callbacks to support running callbacks on executors

[ramitkataria]

Introduce a generic Callback model that can be used for Deadline callbacks for now and hopefully the on_*_callbacks for Dags and Tasks in the future.

Introduce a generic Callback model to run callbacks on executor workers. To be used for Deadline callbacks and the on_*_callbacks for Dags and Tasks that run on the Dag Processor. Implementation of some of the functions will be added in another PR so that this PR can be focused on the DB model and the Workload definition.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[ramitkataria]

Relevant email thread

I've been working on making Deadline callbacks work on the executor in addition to the Triggerer. That means introducing a new Workload type for the executor. I didn't want to tie it specifically to Deadlines and realized it would be a good idea to introduce generic Callbacks that the new workload can reference.

This is an initial PR for adding a new Callback table and refactoring existing Deadline callbacks that run on the Triggerer to use these instead: Ideally, once AIP-92 is in progress, these callbacks can be used or all the on_*_callbacks as well. I want to ensure that these are generic enough for additional subclasses that would implement them. So I'm looking for feedback/comments to align on a common definition for the new Workload type and the Callback model definition.

[ramitkataria]

I've highlighted the areas of this PR I want to focus on and get everyone working on AIP-92 to agree on. Everything else is just refactoring to make existing functionality work. I'll work on unit tests once we agree on how these should work

[vikramkoka]

Relevant email thread

I've been working on making Deadline callbacks work on the executor in addition to the Triggerer. That means introducing a new Workload type for the executor. I didn't want to tie it specifically to Deadlines and realized it would be a good idea to introduce generic Callbacks that the new workload can reference.

This is an initial PR for adding a new Callback table and refactoring existing Deadline callbacks that run on the Triggerer to use these instead: Ideally, once AIP-92 is in progress, these callbacks can be used or all the on_*_callbacks as well. I want to ensure that these are generic enough for additional subclasses that would implement them. So I'm looking for feedback/comments to align on a common definition for the new Workload type and the Callback model definition.

I am quite surprised by this and I guess my core question here is almost "why?"
There seems to be a very large overlap between a sync callback and a task here.

For now, keeping it focused on "sync deadline alert callbacks", this seems be a lot of code replication for unclear benefit.

[o-nikolas]

Relevant email thread
I've been working on making Deadline callbacks work on the executor in addition to the Triggerer. That means introducing a new Workload type for the executor. I didn't want to tie it specifically to Deadlines and realized it would be a good idea to introduce generic Callbacks that the new workload can reference.
This is an initial PR for adding a new Callback table and refactoring existing Deadline callbacks that run on the Triggerer to use these instead: Ideally, once AIP-92 is in progress, these callbacks can be used or all the on_*_callbacks as well. I want to ensure that these are generic enough for additional subclasses that would implement them. So I'm looking for feedback/comments to align on a common definition for the new Workload type and the Callback model definition.

I am quite surprised by this and I guess my core question here is almost "why?" There seems to be a very large overlap between a sync callback and a task here.

For now, keeping it focused on "sync deadline alert callbacks", this seems be a lot of code replication for unclear benefit.

Hey @vikramkoka! Thanks for weighing in, here is more context:

Basically the story goes:

1. Months ago on a dev email list/Thursday dev call discussion the community decided that async Deadlines Callbacks shall run in the Triggerer and that sync callbacks should run on the Executor workers (see your summary of the dev call here, screenshot of the relevant piece below [1] and a follow up from Ramit here).
2. To run things on executors, now with task API, you need to submit a Workload to them. Reusing the Task Workload type does not make much sense since it requires different inputs and runtime behaviour than callbacks, so we are creating a new Workload type.
3. While creating that new Workload type, instead of making it specific to Deadline callbacks, we're making it generic enough to run any callback. This means that we can eventually move the on_success/on_failure callbacks to run on the executors in the same way, relieving the Dag Parser of this duty and having all user code executed, rather sensibly, on the executors.
4. However, AIP-92 has also been in discussion lately, and there they have a proposal to create a separate process to run the on_success/on_failure callbacks [2], which is an approach we thought we ruled out in 1). So we're looking to reevaluate and discuss to orient back onto a single path for callback execution. And whatever path we decide, is what we'll use for sync callbacks for Deadlines. Which will be delivered for 3.2 as discussed separately.

[1]

[2]

[ramitkataria]

I've removed the refactoring changes for deadline and just focused on the DB related changes for callbacks to keep this PR focused on the major change. There will be followup PRs to refactor deadlines to use TriggererCallback as well as implementing support for Executor callbacks