ephraimbuddy opened a new pull request, #76303:
URL: https://github.com/apache/airflow/pull/76303
Dag success and failure callbacks, and the task failure and retry callbacks
and emails that a task cannot run itself (for example after it was killed
externally or stopped heartbeating), only run inside a Dag processor parse.
That ties them to the hosts that parse the bundle rather than the executors and
queues their tasks use, runs them without their team's Connections and
Variables, and loses them without a trace if the processor dies after claiming
them.
This adds an opt-in `[callbacks] route_dag_processor_callbacks` setting, off
by default. With it on, the scheduler routes these callbacks to executors as
callback workloads, which executors already run for deadline callbacks:
- A task's callbacks run on the task's executor and queue. Dag callbacks,
and callbacks of a task that never started (such as one failed while stuck in
the queue), run on the team's default executor and queue. The worker parses the
Dag file at the Dag run's bundle version.
- Callbacks take at most half the free executor slots in a loop, globally
and per executor, so a burst of them can't keep tasks from running.
`[callbacks] max_routed_callbacks_per_loop` caps each loop.
- Every executor must be able to run callbacks, or the scheduler doesn't
start. A callback no executor can run, such as one of a Lang-SDK Dag, fails
with the reason instead of going back to a Dag processor.
- A callback that doesn't start by its deadline is failed and routed again
under a new id, up to three tries, so the token issued for the earlier try
can't start it. One whose worker doesn't report it finished within `[callbacks]
routed_callback_execution_timeout` plus a grace period is failed.
- Turning routing off hands callbacks not yet routed back to the Dag
processor. Routed ones still run on their executors.
Other changes:
- Execution API (unreleased version 2027-02-28): `GET
/callbacks/{callback_id}/context`, `GET
/callbacks/{callback_id}/previous-successful-dagrun` and `PATCH
/callbacks/{callback_id}/state`. An attempt to run a callback can repeat its
own token exchange, as after a lost Celery acknowledgement. Callback execution
tokens carry `dag_id`, `run_id` and `bundle_name`, so a callback's team comes
from its signed bundle name.
- Migration 0154 adds an index for looking callbacks up by type, state and
fetch method.
- New metrics: `scheduler.routed_callbacks.reaped` and `callback.timeouts`.
- Executors gain `release_callback`, which drops a callback the scheduler
settled without an executor event. LocalExecutor, Celery, ECS, Batch and Lambda
implement it: ECS stops a task whose `RunTask` call was still in flight, and
Lambda keeps acknowledging late SQS results.
related: #44354
related: #62887
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 5.5), Codex (GPT6)
Generated-by: Claude Code (Opus 5.5) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
# NOTE: This will be broken down into smaller PRs later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
