Executor Synchronous callback workload #61153

ferruzzi · 2026-01-27T23:13:04Z

Add support for the Callback workload to be run in the executors. Other executors will need to be updated before they can support the workload, but I tried to make it as non-invasive as I could.

This is the bulk of the work required to allow synchronous callbacks to be used in DeadlineAlerts. For example, this now works in LocalExecutor:

with DAG(
    dag_id="sync_deadline",
    deadline=DeadlineAlert(
        reference=DeadlineReference.FIXED_DATETIME(datetime(1980, 8, 10, 2)),
        interval=timedelta(0),
        callback=SyncCallback(
            SlackWebhookNotifier,
            {"text": "Sync Callback; Alert should trigger immediately!"},
        )
    )
):
    EmptyOperator(task_id='empty_task')

Co-author: Builds on work handed off by @seanghaeli and research from @ramitkataria; if I did this right then they should be getting co-author credits, I think?

Was generative AI tooling used to co-author this PR?

[ x ] Yes (please specify the tool below)
Cline (Claude Sonnet 4.5) was used for debugging and suggesting some unit test edge cases.

Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
When adding dependency, check compliance with the ASF 3rd Party License Policy.
For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

…eryExecutor Add support for the Callback workload to be run in the executors. Other executors will need to be updated before the can support the workload, but I tried to make it as non-invasive as I could.

airflow-core/src/airflow/executors/base_executor.py

airflow-core/src/airflow/executors/local_executor.py

o-nikolas · 2026-01-27T23:43:38Z

airflow-core/src/airflow/executors/local_executor.py

+            key = workload.callback.id
+            try:
+                _execute_callback(log, workload, team_conf)
+                output.put((key, TaskInstanceState.SUCCESS, None))


Mostly just curious: We still use TaskInstanceState here even those these are callbacks?

I chewed this one a bit.... Its' either a SUCCESS or FAIL and I felt heavy handed adding yet another state machine enum to repeat those values. I have an idea I'll include in the next revision and see what you think.

o-nikolas · 2026-01-27T23:53:10Z

airflow-core/src/airflow/jobs/scheduler_job_runner.py

+            # Find the appropriate executor
+            executor = None
+            if executor_name:
+                # Find executor by name - try multiple matching strategies
+                for exec in self.job.executors:
+                    # Match by class name (e.g., "CeleryExecutor")
+                    if exec.__class__.__name__ == executor_name:
+                        executor = exec
+                        break
+                    # Match by executor name attribute if available
+                    if hasattr(exec, "name") and exec.name and str(exec.name) == executor_name:
+                        executor = exec
+                        break
+                    # Match by executor name attribute if available
+                    if hasattr(exec, "executor_name") and exec.executor_name == executor_name:
+                        executor = exec
+                        break
+
+            # Default to first executor if no specific executor found
+            if executor is None:
+                executor = self.job.executors[0] if self.job.executors else None
+
+            if executor is None:
+                self.log.warning("No executor available for callback %s", callback.id)
+                continue


This is also missing multi-team logic which we need to stay up to date with at this point. It also is duplicating a lot of the work in _try_to_load_executor which is made to do exactly this kind of lookup. I think it's going to save you a bunch of effort and future maintenance to update _try_to_load_executor to support workloads generally instead of just ti (basically exactly the type of coding you did in the base executor and local executor changes).

o-nikolas · 2026-01-27T23:53:47Z

airflow-core/src/airflow/jobs/scheduler_job_runner.py

+                self.log.warning("No executor available for callback %s", callback.id)
+                continue
+
+            executor_to_callbacks[executor].append(callback)


Similar to the above, there is already a _executor_to_tis which is doing exactly this but for tis, could be generalized.

o-nikolas · 2026-01-28T00:01:19Z

airflow-core/src/airflow/jobs/scheduler_job_runner.py

+            .where(ExecutorCallback.type == CallbackType.EXECUTOR)
+            .where(ExecutorCallback.state == CallbackState.QUEUED)
+            .order_by(ExecutorCallback.priority_weight.desc())
+            .limit(conf.getint("scheduler", "max_callback_workloads_per_loop", fallback=100))


Here and down below in the final loop over executors/workloads we're just queueing a static amount each time. But it is the schedulers responsibility now (in the world of multiple executors and now multi-team) to ensure we don't ever schedule more tasks (now, workloads) than we have executor slots for. You can see how we do this math for tasks currently here:

airflow/airflow-core/src/airflow/jobs/scheduler_job_runner.py

Lines 854 to 871 in 056e24e

# The user can either request a certain number of tis to schedule per main scheduler loop (default

# is non-zero). If that value has been set to zero, that means use the value of core.parallelism (or

# however many free slots are left). core.parallelism represents the max number of running TIs per

# scheduler. Historically this value was stored in the executor, who's job it was to control/enforce

# it. However, with multiple executors, any of which can run up to core.parallelism TIs individually,

# we need to make sure in the scheduler now that we don't schedule more than core.parallelism totally

# across all executors.

num_occupied_slots = sum([executor.slots_occupied for executor in self.job.executors])

parallelism = conf.getint("core", "parallelism")

if self.job.max_tis_per_query == 0:

max_tis = parallelism - num_occupied_slots

else:

max_tis = min(self.job.max_tis_per_query, parallelism - num_occupied_slots)

if max_tis <= 0:

self.log.debug("max_tis query size is less than or equal to zero. No query will be performed!")

return 0

queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)

We need to ensure that math now includes callbacks because they also take up worker slots.

I think this will work for now, as long as this method is always called before the critical section. Since callbacks will increase occupied slots in the executors which should be taken into account in the critical section. BUT this code here needs to ensure it doesn't over subscribe the executors. So some similar logic to the critical section needs to be done here. E.g. we're taking a flat 100 here (by default anyway) but there may only be 20 free executor slots.

Ah. I think I see what you are saying, I missed that. I may need to chat offline to clarify, but I'll look into it. Thanks for catching that.

This should be addressed if I understood correctly.

o-nikolas · 2026-01-28T00:03:02Z

airflow-core/src/airflow/jobs/scheduler_job_runner.py

+            if callback:
+                # Note: We receive TaskInstanceState from executor (SUCCESS/FAILED) but convert to CallbackState here.
+                # This is intentional - executor layer uses generic completion states, scheduler converts to proper types.
+                if state == TaskInstanceState.SUCCESS:


I think this is fine for now, but would be cool if Callbacks were fully first class citizens in executors. Including executors reporting the right state back.

This is going to fall under the same fix as #61153 (comment)

I was on the fence about this but you've convinced me.... Next revision will have a more generic WorkloadState and WorkloadKey. Code is done, just testing it out now.

That's up now, let me know what you think

…fixes

jscheffl · 2026-01-29T21:01:11Z

providers/celery/src/airflow/providers/celery/executors/celery_executor.py

    def _process_workloads(self, workloads: Sequence[workloads.All]) -> None:
        # Airflow V3 version -- have to delay imports until we know we are on v3
-        from airflow.executors.workloads import ExecuteTask
+        from airflow.executors.workloads import ExecuteCallback, ExecuteTask


The ExecuteCallback was added leter than 3.0.0, so if you import here it will break in core versions <3.2.0?

Can you either make a (very) late or lazy import such that no upgrade dependency to airflow core is generated and that it is still working with 3.0.0?

Yeah, for sure, good catch. I can do a soft test like :

if isinstance(workload, ExecuteTask): # handle task elif type(workload).__name__ == "ExecuteCallback": # Import only if callback found from airflow.executors.workloads import ExecuteCallback

or a hard gate like

def _process_workloads(self, workloads): from airflow.executors.workloads import ExecuteTask if AIRFLOW_V_3_2_PLUS: from airflow.executors.workloads import ExecuteCallback

The first option feels better, but the second looks more explicit. Any preference?

seanghaeli and others added 2 commits January 27, 2026 11:21

first pass implementation of executor support for sync callbacks

24fd91a

Synchronous callback support for BaseExecutor, LocalExecutor, and Cel…

97e09db

…eryExecutor Add support for the Callback workload to be run in the executors. Other executors will need to be updated before the can support the workload, but I tried to make it as non-invasive as I could.

ferruzzi requested a review from vincbeck January 27, 2026 23:13

ferruzzi requested review from XD-DENG, amoghrajesh, ashb, dheerajturaga, hussein-awala, kaxil, o-nikolas and pierrejeambrun as code owners January 27, 2026 23:13

boring-cyborg bot added area:deadline-alerts AIP-86 (former AIP-57) area:Executors-core LocalExecutor & SequentialExecutor area:providers area:Scheduler including HA (high availability) scheduler area:task-sdk provider:celery labels Jan 27, 2026

o-nikolas reviewed Jan 28, 2026

View reviewed changes

ferruzzi added 2 commits January 28, 2026 12:24

Create a unified WorkflowState and WorkflowKey; added mypy type hint …

5bc74f4

…fixes

First batch Niko fixes

cfb3459

ferruzzi requested review from jedcunningham and jscheffl as code owners January 29, 2026 20:35

deovercommentification of hand-off notes

d3cf1ce

jscheffl reviewed Jan 29, 2026

View reviewed changes

ferruzzi added 4 commits January 29, 2026 15:05

CI fixes and minor type-related cleanup

5f63b2e

Niko suggestions round 2 and Ci/prek fixes

6876869

CI and MyPy fixes

2247c53

CI and MyPy fixes

bd351a9

	# The user can either request a certain number of tis to schedule per main scheduler loop (default
	# is non-zero). If that value has been set to zero, that means use the value of core.parallelism (or
	# however many free slots are left). core.parallelism represents the max number of running TIs per
	# scheduler. Historically this value was stored in the executor, who's job it was to control/enforce
	# it. However, with multiple executors, any of which can run up to core.parallelism TIs individually,
	# we need to make sure in the scheduler now that we don't schedule more than core.parallelism totally
	# across all executors.
	num_occupied_slots = sum([executor.slots_occupied for executor in self.job.executors])
	parallelism = conf.getint("core", "parallelism")
	if self.job.max_tis_per_query == 0:
	max_tis = parallelism - num_occupied_slots
	else:
	max_tis = min(self.job.max_tis_per_query, parallelism - num_occupied_slots)
	if max_tis <= 0:
	self.log.debug("max_tis query size is less than or equal to zero. No query will be performed!")
	return 0

	queued_tis = self._executable_task_instances_to_queued(max_tis, session=session)

Executor Synchronous callback workload #61153

Are you sure you want to change the base?

Executor Synchronous callback workload #61153

Conversation

ferruzzi commented Jan 27, 2026

Was generative AI tooling used to co-author this PR?

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants