[Leondon9]

This makes GlueJobOperator's stop_job_run_on_kill work on the deferrable path: when a deferred Glue task is cleared, marked failed or marked success, the Glue job run is stopped instead of being left running.

Motivation

stop_job_run_on_kill was only honoured by the operator's synchronous on_kill, which does not run for a deferred task (the task lives in the triggerer, not on a worker). So a cleared deferrable Glue task left its job run going. EmrContainerTrigger already stops its job from on_kill(); this brings Glue in line.

Behaviour

• Opt-in and unchanged by default (stop_job_run_on_kill=False).
• GlueJobCompleteTrigger.on_kill() stops the run. It uses the async client, so client creation stays off the triggerer event loop, and it logs batch_stop_job_run Errors instead of ignoring them. Requires Airflow 3.3+, where the triggerer calls BaseTrigger.on_kill(); on older Airflow it is inert, same as EmrContainerTrigger.
• Clear starts a fresh run. on_kill fires on every user action and cannot tell a clear (which re-runs the task) from mark failed/success. With durable on (the 3.3+ default), the cleared task's new try would reconnect to the run on_kill is stopping and then fail on STOPPED. So when stop_job_run_on_kill=True and durable=True in deferrable mode on 3.3+, a retried try stops the previous run itself, waits for it to reach a terminal state, then submits a new run. The wait matters because a STOPPING run still counts toward MaxConcurrentRuns, so submitting earlier would hit ConcurrentRunsExceededException.
  • If the previous-run lookup fails, it falls back to submitting, same as the existing durable path.
  • If the run is still not terminal after 10 minutes, the try fails with GlueJobRunStopTimeoutError rather than starting a second run alongside it.
  • Without durable the job arguments carry no task UUID, so the previous run cannot be found; the retry submits a fresh run straight away and relies on on_kill to stop the old one. With MaxConcurrentRuns=1 that submission can still hit ConcurrentRunsExceededException, as it does today.
• durable and its default are untouched, and stop_job_run_on_kill=False keeps the existing reconnect behaviour.
• Released 3.3.x cores also call on_kill() when a trigger is reassigned to another triggerer (fixed on main in Skip on_kill() for triggers reassigned to another triggerer #73454). The docstring notes this.

Synchronous mode is unchanged.

Testing

• Trigger: on_kill stops the run when enabled, logs on Errors, and does nothing when disabled; stop_job_run_on_kill serializes.
• Operator: the flag reaches the trigger; a deferrable retry with a live previous run stops it, waits through STOPPING, and submits a fresh run, including when on_kill already stopped it; the try fails without submitting if the run never stops; without durable the retry submits fresh without a lookup.
• Glue trigger and operator unit tests pass (143). prek pre-commit hooks and provider mypy (breeze run mypy) on the changed files pass.
• Not run against a live AWS Glue job. The local amazon/aws-glue-libs image does not provide Glue's control-plane API.

---

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

• Yes — Claude Code (Opus 4.8, Opus 5.5)

Generated-by: Claude Code (Opus 4.8, Opus 5.5) following the guidelines

🤖 Generated with Claude Code

[vincbeck]

@seanghaeli @ramitkataria

[seanghaeli]

On my machine I'm seeing a problem arise with this PR when a deferred task is cleared. clear instructs the task to run again, and on 3.3+ the re-run looks for an existing Glue job from the same task. It finds the old one still running, and decides to just wait on that one. But clearing also triggers this PR's "on kill" step, which stops that same old Glue job. So the re-run ends up waiting on a stopped job and fails with STOPPED. On main, the re-run reattached and succeeded.

Could the re-run start a fresh Glue job when stop_job_run_on_kill=True, instead of reattaching?

[Leondon9]

On my machine I'm seeing a problem arise with this PR when a deferred task is cleared. clear instructs the task to run again, and on 3.3+ the re-run looks for an existing Glue job from the same task. It finds the old one still running, and decides to just wait on that one. But clearing also triggers this PR's "on kill" step, which stops that same old Glue job. So the re-run ends up waiting on a stopped job and fails with STOPPED. On main, the re-run reattached and succeeded.

Could the re-run start a fresh Glue job when stop_job_run_on_kill=True, instead of reattaching?

Thanks for reproducing this, you're right. on_kill fires on any user action and can't tell a clear from mark failed/success, so the re-run reattached to the run it was stopping.
Fixed in db0041c: with stop_job_run_on_kill=True in deferrable mode, a retried try no longer reattaches. It stops the previous run itself, waits for it to reach a terminal state (a STOPPING run still counts toward MaxConcurrentRuns), then submits a fresh one.
If the run doesn't stop within 10 minutes, the try fails instead of starting a second run.
durable and its default are unchanged, so stop_job_run_on_kill=False still reattaches as before.
Would you mind re-running your clear scenario?

[o-nikolas]

With durable=False (or the deprecated resume_glue_job_on_retry=False), _build_script_args never adds --airflow_task_uuid, so script_args[self.TASK_UUID_ARG] raises KeyError on every retry. Could this condition also check self.durable? Without the tag the previous run can't be found anyway. A test for that combination would help too.

[Leondon9]

Good catch, thanks. Added self.durable to the condition in ff4f01f, with a test for durable=False, stop_job_run_on_kill=True: no lookup, no stop, just a fresh submission. Without durable the trigger's on_kill still stops the old run, but the new submission can race it and hit ConcurrentRunsExceededException with MaxConcurrentRuns=1, as it does today. I noted that in the description.