seanmuth commented on issue #60858: URL: https://github.com/apache/airflow/issues/60858#issuecomment-6177199669
Update: @ashb and I went a few rounds on this internally after his comment above, so that comment is a bit behind where we ended up. Below is the design we converged on. It also replaces the "signal kill" proposal in my earlier comment. @eladkal @potiuk, we'd like your weigh-in before anyone starts on a PR. Two things changed from my earlier proposal: - **No exit-code inference.** Keying on a negative exit code is Python-specific, and with multiple language SDKs that doesn't hold. It also isn't correct today: the fail-closed path in the runner exits `1` *after* `finalize()` has already run callbacks. - **Nothing extra runs in the API server.** Listeners (OpenLineage especially) can block, and the API server has no executor. `send_callback()` is a property of the executor and is meant to be overridable, so routing has to happen in the scheduler. We also ruled out the supervisor spawning a new process to run the callback. If the task was OOM-killed, a second process in the same pod can just get OOM-killed again. ### Design 1. **Task SDK.** The runner sends a `FinalizeStarted` message to the supervisor when it enters [`finalize()`](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/task-sdk/src/airflow/sdk/execution_time/task_runner.py#L2390). If the task process exits with no terminal state, no `FinalizeStarted`, and no kill the supervisor started itself (overtime, heartbeat loss, server-requested), the supervisor reports `FAILED`, or `UP_FOR_RETRY` if `should_retry`, with `failure_handling_ran=False`. Sending `UP_FOR_RETRY` explicitly also closes the [`STATES_SENT_DIRECTLY`](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L309) gap. 2. **Execution API.** Add `failure_handling_ran: bool = True` to the terminal and retry payloads, so old SDKs behave as today. When it's `False`, the API server builds the same `TaskCallbackRequest` the scheduler builds today (plus the email request if configured) and inserts it into the `callback` table as a `DbCallbackRequest` with `state=PENDING`, in the same transaction as the state change. It's built before `prepare_db_for_next_try`, so it describes the failed try. DB writes only, no listeners, no executor. 3. **Scheduler.** A new loop, modelled on the [`ExecutorCallback` router](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1334), selects `PENDING` rows with `SKIP LOCKED`. For each row it calls [`send_callback()`](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/airflow-core/src/airflow/executors/base_executor.py#L834) on the TI's executor, calls `notify_failure()` (so `on_task_instance_failed` and the OL FAIL event fire where they already fire for the existing scheduler paths), and deletes the row in the same transaction. 4. **Dag processor.** Its [callback fetch](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/airflow-core/src/airflow/dag_processing/manager.py#L740) gets `.where(state IS NULL)`, so it only picks up rows the scheduler has routed. Callback execution itself doesn't change. **No migration.** The `callback` table already has a `state` column, and dag processor rows always have `state=NULL` today. I checked the other readers of the table (the executor callback router, `airflow db clean`, the deadline and trigger joins) and none of them are affected by `PENDING` dag processor rows. Before any of that, a no-op refactor: pull the callback-building block that's duplicated in the [executor-mismatch](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1743) and [heartbeat-purge](https://github.com/apache/airflow/blob/d189eac87c1f8bed96efdbf0a89bc6d7784da45c/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L4109) paths into one helper, so the API server and both scheduler paths build identical requests. ### Delivery count (re: Ash's question above) - **Callback:** at most once per failed try. If `finalize()` started, the supervisor doesn't flag the try, even if `finalize()` died partway through. Each pending row is claimed under a row lock and deleted in the same transaction that hands it on. With the default `DatabaseCallbackSink` that hand-off is an insert in the same transaction, so in practice it's exactly once. - **Listener / OL FAIL:** once in the normal case. The only window for a duplicate is the scheduler dying after `notify_failure()` and before commit, which is the same exposure the existing scheduler paths already have. If OL needs strict at-most-once we can move the listener call after commit, which flips that window to "missed" instead. ### Other properties - **Latency:** one scheduler loop. Today it's the heartbeat timeout for the retry case, and never for the `FAILED` case. - **If the scheduler is down,** pending rows wait. If the supervisor dies too, the existing scheduler paths still cover it. - **Rolling upgrade:** an old dag processor could pick up a `PENDING` row directly. The callback still runs, it just skips executor routing, which is still better than today. - **Release:** the Execution API change means a Cadwyn version bump, so this targets the next minor. ### Open questions 1. Should the routing loop get its own timer, or run alongside the `ExecutorCallback` router? 2. Should listener calls on the scheduler paths (existing and new) move off the main loop as a follow-up? 🤖 Generated with [Claude Code](https://claude.com/claude-code) — Claude Opus 5.5 -- 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]
