Cancel the Redshift statement when a user kills the deferred task

[steveahnahn]

Problem

• RedshiftDataOperator.on_kill cancels the running statement, but after defer() the worker is gone and it can never run.
• RedshiftDataTrigger had no on_kill (base default is a no op), so killing a deferred task orphans the statement: RPU hours keep billing on Serverless, a WLM slot stays held on a provisioned cluster.
• Sibling triggers already cancel on user kill: EMR here, BigQuery/Dataproc/Dataflow in the Google provider.

Change

• Add on_kill to RedshiftDataTrigger: cancels via the hook's async connection (cancel_statement).
• CancelStatement returns {"Status": bool}; a declined cancel (statement already finished) logs a warning instead of claiming success.
• cancel_on_kill flag (default True) on operator and trigger; operator threads it into the trigger at the defer site.
• Best effort: cancel failures are logged and swallowed. Fires only on user kill, never on triggerer restart, redistribution, or timeout.

Live verification

1. Deferrable operator submitted a statement (STARTED); task deferred, trigger polling via the real aiobotocore client.

2. Marked the task failed from the UI.

3. Triggerer log, no exception escaping:

Trigger cancelled by user action, invoking on_kill
Cancelling Redshift statement 93e26017-b365-4c9f-9c58-9d387af5e6ea.
Redshift statement 93e26017-b365-4c9f-9c58-9d387af5e6ea cancelled.

4. DescribeStatement returned Status: ABORTED. Without the change the statement stays STARTED (base on_kill is a no op).

Tests

• Trigger: cancel on kill (parametrized over Status True/False with the real response shape), disabled/missing id no ops, error swallowing, serialization of the flag.
• Operator: cancel_on_kill=False guard, flag threaded into the trigger on defer (both True and False).
• Every new assertion fails without the source change.

---

Was generative AI tooling used to co-author this PR?

• Yes, Claude Code (Fable 5)

Generated-by: Claude Code (Fable 5) following the guidelines

[potiuk]

Good catch on a gap with a direct running cost — after defer() the worker is gone, so the operator's on_kill could never fire, and an orphaned statement keeps billing RPU hours on Serverless or holds a WLM slot on a provisioned cluster.

The part I care most about is where the cancellation lives, and it is in the right place. BaseTrigger's own docstring warns that putting external job cancellation in cleanup() would cancel in-flight work on every triggerer restart or rolling deploy; this uses on_kill(), and the triggerer only invokes that on an explicit user action — the timeout path raises without calling it. So a deploy will not take out live queries. It is also bounded by _ON_CANCEL_TIMEOUT, so a slow cancel cannot wedge the triggerer.

Adding cancel_on_kill to serialize() matters more than it looks: without it the flag would be silently dropped crossing into the triggerer, and the opt-out would appear to work while doing nothing.

Handling CancelStatement returning Status=False is a nice detail — reporting "declined, may have already finished" rather than claiming a successful cancel is the difference between a log you can trust and one you cannot. Swallowing cancel errors with log.exception is right too; a failed cancel must not block the kill.

Worth being explicit in the release notes that killing a deferred task now cancels the statement where it previously orphaned it. That is the intended fix, and cancel_on_kill=False is the escape hatch, but it does change what a kill does for anyone who had built expectations around the old behaviour.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting