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.

o-nikolas · 2026-01-27T23:25:08Z

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

+        remaining_slots = open_slots - len(workloads_to_schedule)
+        if remaining_slots and self.queued_tasks:


nit: remaining_slots is only used once. You could just put something like open_slots > len(workloads_to_schedule) in the if expression

o-nikolas · 2026-01-27T23:26:23Z

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

+
+        remaining_slots = open_slots - len(workloads_to_schedule)
+        if remaining_slots and self.queued_tasks:
+            sorted_tasks = sorted(


Why didn't you use the existing order_queued_tasks_by_priority() method?

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

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

+            try:
+                self._process_workloads(workload_list)
+            except AttributeError as e:
+                if any(isinstance(workload, workloads.ExecuteCallback) for workload in workload_list):


If we know exactly how to check for the unsupported use case, why don't we just check before trying to call __process_workloads()? Also, we can check much earlier in the queueing of workloads because we can check the supports_callback attr?

o-nikolas · 2026-01-27T23:42:18Z

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

    TaskInstanceStateType = tuple[workloads.TaskInstance, TaskInstanceState, Exception | None]


+def _get_executor_process_title_prefix(team_name: str | None) -> str:


These multi team related changes probably shouldn't be showing up in this diff right?

Yeah, I might have botched the rebase before I published this. I'll try to extricate those changes.

_execute_callback() is the third time duplicating that exact pattern, so I moved it to a helper.

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?

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.

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.

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Executor Synchronous callback workload #61153

Executor Synchronous callback workload #61153

ferruzzi commented Jan 27, 2026

Uh oh!

o-nikolas Jan 27, 2026

Uh oh!

o-nikolas Jan 27, 2026

Uh oh!

o-nikolas Jan 27, 2026 •

edited

Loading

Uh oh!

o-nikolas Jan 27, 2026

Uh oh!

ferruzzi Jan 28, 2026

Uh oh!

ferruzzi Jan 28, 2026 •

edited

Loading

Uh oh!

o-nikolas Jan 27, 2026

Uh oh!

o-nikolas Jan 27, 2026

Uh oh!

o-nikolas Jan 27, 2026

Uh oh!

o-nikolas Jan 28, 2026

Uh oh!

o-nikolas Jan 28, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		remaining_slots = open_slots - len(workloads_to_schedule)
		if remaining_slots and self.queued_tasks:

		TaskInstanceStateType = tuple[workloads.TaskInstance, TaskInstanceState, Exception \| None]


		def _get_executor_process_title_prefix(team_name: str \| None) -> str:

	# 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

o-nikolas Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

ferruzzi Jan 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

3 participants

o-nikolas Jan 27, 2026 •

edited

Loading

ferruzzi Jan 28, 2026 •

edited

Loading