[kaxil]

When a triggerer's heartbeat lapses for longer than triggerer_health_check_threshold, Trigger.assign_unassigned hands its triggers to another triggerer. The old triggerer is often still alive: the runner-silence watchdog deliberately skips heartbeats when the runner subprocess stops responding, and a long GC pause, a stopped container or a network partition do the same. When it recovers, TriggerRunnerSupervisor.update_triggers sees those triggers missing from its assignment list and queues them for cancellation with the user-action message, so run_trigger calls on_kill() on every one of them. Nothing in that path distinguishes "the row was deleted because a user cleared the task" from "the row was reassigned".

For any trigger that cancels remote work from on_kill() this cancels a job the new owner has just started polling. The batch triggers in the Anthropic and Common AI providers, and the EMR, Redshift Data, Databricks, Dataproc and BigQuery triggers all do. The BaseTrigger.on_kill docstring and the deferring docs both promise it does not fire on redistribution, so this is a contract violation, present since on_kill() shipped in 3.3.0 (#65590). The cancel_triggers docstring claimed redistribution "goes through a separate path"; there was no such path.

Solution

Before queueing a cancel, the supervisor asks the DB which of the departed ids still have a trigger row and who owns them now (Trigger.fetch_assignments, behind an overridable fetch_trigger_assignments hook that owns its own session so update_triggers stays DB-free for subclasses without a metadata DB). A row that survives after leaving our list was reassigned, so those ids go to a new releasing_triggers set and reach the runner as TriggerStateSync.to_release; the runner cancels those asyncio tasks with a distinct sentinel, and run_trigger drops the trigger without on_kill(). Only a row that is gone, which is what a clear or mark-success/failed produces, takes the existing user-action path. The classification runs once per departed id, since ids already queued are excluded from the next loop's difference. The new field defaults to an empty set, so existing TriggerStateSync producers are unaffected.

Why in the supervisor rather than the runner. The runner subprocess has no DB access; the supervisor already owns the assignment query and the per-loop create_session() in build_trigger_workloads, so the one extra SELECT id ... WHERE id IN (...) sits next to it, and the runner keeps its existing message-based contract.

Reproduction

Two triggerers started with --queues afl297 and triggerer.queues_enabled=True (so the standalone triggerer stays out of it), a task on that queue deferring to a trigger whose on_kill() appends a line to a file, then SIGSTOP on the owning triggerer and its runner child until the trigger row moved, then SIGCONT.

	Old owner after SIGCONT	on_kill() marker
main	Trigger cancelled by user action, invoking on_kill	written by the old owner's runner while the row already belonged to the new owner and the task was still deferred
this PR	Triggers were reassigned to another triggerer, releasing them without on_kill new_owners={3: 8} and, in the runner, Trigger reassigned to another triggerer, dropping it without on_kill	not written

With this PR, marking the deferred task failed through the REST API still produced the marker from the new owner's runner within seconds, so the user-action path is intact.

Known gap

Row existence proves the row was reassigned, not that the new owner has started it. If a user clears the task in the window after the old owner classified the trigger as reassigned and before the new owner fetched its workload, neither side calls on_kill(): the old owner released, and the new owner skips a trigger whose task instance is gone. This is the same outcome main has when the old owner has actually died, so on_kill() on a user action was already best-effort while a failover is in flight. The fix errs toward leaving remote work running rather than cancelling work another triggerer is polling.

---

• 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, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[amoghrajesh]

Good fix, I have some nits, otherwise looks good.

[flo076]

Hello @kaxil ,

Thanks for this fix. Is a backport to the Airflow 3.3.x release line planned, and if so, which version would include it? We observed explicit BigQuery CancelJob calls while a BigQueryInsertJobOperator task was deferred on Airflow 3.3.1. We have not confirmed that trigger reassignment caused our incident, but the behavior described in this PR appears relevant. The fix is not present in the 3.3.1 or 3.3.2 tags.

[github-actions]

Backport failed to create: v3-4-test. View the failure log Run details

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
❌	v3-4-test

You can attempt to backport this manually by running:

cherry_picker eebf253 v3-4-test

This should apply the commit to the v3-4-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.

After you have resolved the conflicts, you can continue the backport process by running:

cherry_picker --continue

If you don't have cherry-picker installed, see the installation guide.

[kaxil]

Hello @kaxil ,

Thanks for this fix. Is a backport to the Airflow 3.3.x release line planned, and if so, which version would include it? We observed explicit BigQuery CancelJob calls while a BigQueryInsertJobOperator task was deferred on Airflow 3.3.1. We have not confirmed that trigger reassignment caused our incident, but the behavior described in this PR appears relevant. The fix is not present in the 3.3.1 or 3.3.2 tags.

Unfortunately, it will only be part of 3.4.0 since 3.4.0rc1 is about to be created

[flo076]

Thanks for clarifying. The impact for us is broader than BigQuery. We have observed unexpected interruptions affecting multiple deferrable GCP operators, including BigQuery and Dataproc. In the BigQuery case, audit logs confirm explicit CancelJob calls while the task was deferred; two successive attempts of the same task ended in cancellation, so a retry did not reliably recover it.

We cannot yet prove that trigger reassignment caused every incident. Nevertheless, the core failure mode fixed by this PR affects multiple remote-job triggers, and makes deferrable mode effectively unsafe for our affected workloads on Airflow 3.3.1. Falling back to non-deferrable execution consumes workers for the duration of these jobs.

Given this operational impact, could a 3.3.x backport be reconsidered? If that is not possible, is there a supported mitigation for managed environments that cannot wait 3.4.X ?