steveahnahn opened a new pull request, #69586: URL: https://github.com/apache/airflow/pull/69586
Killing a deferred Dataflow task (mark-failed, clear, or mark-succeeded) currently leaves the Dataflow job running in Google Cloud: `TemplateJobStartTrigger` has no `on_kill`, so the user-kill cancellation that the sibling BigQuery and Dataproc triggers already perform never happens — a streaming pipeline keeps running and billing until someone notices in the Cloud console. Before the Airflow 3.3 trigger-cancellation redesign the legacy `asyncio.CancelledError` path cancelled jobs from inside the trigger, so for deferrable users this is also a behavior regression relative to earlier releases. This implements `on_kill` on `TemplateJobStartTrigger` following the merged BigQuery / Dataproc / EMR (#64770) pattern: guard on `cancel_on_kill` / `job_id` / `project_id`, then cancel through the synchronous `DataflowHook` via `sync_to_async`, honoring `drain_pipeline` so streaming jobs are drained rather than cancelled when configured — exactly what the non-deferred kill path does. ### Reviewer notes - Scope is deliberately only `TemplateJobStartTrigger` (used exclusively by `DataflowTemplatedJobStartOperator` and `DataflowStartFlexTemplateOperator`, which own the job). The status/metrics/messages triggers used by sensors are untouched — an observer should not cancel a job it merely watches. - `drain_pipeline` is threaded from the flex-template operator only; `DataflowTemplatedJobStartOperator` does not expose `drain_pipeline` (its synchronous hook is built without it), so its defer site intentionally leaves the trigger default (`False`) to mirror sync behavior exactly. - No operator-level opt-out is added because the synchronous kill path cancels unconditionally too; the trigger-level `cancel_on_kill` parameter exists for parity with the sibling triggers. - The serialize additions (`cancel_on_kill`, `drain_pipeline`) are additive with defaults, so triggers serialized before an upgrade rehydrate fine. - The framework bounds `on_kill` runtime, which can truncate the hook's up-to-`cancel_timeout` wait for cancel *completion* — the cancellation request itself is issued first, so the job is still cancelled server-side. ### Tests - Six new trigger tests: `on_kill` cancels with the right hook construction and arguments; `drain_pipeline` is honored; no-op when `cancel_on_kill=False` / `job_id` missing / `project_id` missing; hook errors are swallowed and logged. On the previous code, 7 tests fail (`on_kill` set plus the serialize update); all 87 tests in the file pass with the change. - Provider mypy clean on both touched files. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes - Claude Code (Opus 4.8) Generated-by: Claude Code (Opus 4.8) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- 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]
