amoghrajesh opened a new issue, #72128:
URL: https://github.com/apache/airflow/issues/72128
### Summary
An operator submits a job to an external system (Spark, Glue, Databricks,
BigQuery) and polls it. If
the task fails without being killed, and has no retries left, nothing
cancels that external job. The
Dag run finishes, the task is red, and the job keeps consuming compute with
nothing in Airflow
tracking it.
This is not specific to durable execution. It affects any operator that
submits work externally and
puts its cancellation logic in `on_kill`, which is currently 54 provider
operators.
### Why `on_kill` does not cover it
`on_kill()` is where operators put cancellation logic, but the task runner
calls it in only two
situations:
- `SIGTERM`, via the handler registered in `task_runner.py`
- `execution_timeout` being exceeded, which raises `AirflowTaskTimeout`
Ordinary failure is neither. The runner catches whatever the task raised and
routes it to
`_handle_current_task_failed`, which picks `FAILED` or `UP_FOR_RETRY` from
the `should_retry` flag the
server sent down before execution started. The operator is never notified,
and there is no distinct
exception for "this was the final attempt".
### Two failure modes, and they need different fixes
These get conflated, but only one of them is solvable in the worker process.
**Mode 1: the task raises and the process is still alive.**
A transient polling error, an expired credential, a network partition, or a
failure in code that runs
after the job was submitted. The worker is healthy and could run cleanup, it
just is not asked to.
This affects every operator that submits externally, durable or not.
`should_retry` is already in the
TI context at this point, so the runner knows whether this was the last
attempt without any new
plumbing.
**Mode 2: the worker is SIGKILLed.**
Out of memory, or the cloud reclaims the worker machine. `SIGKILL` cannot be
trapped, so no in-process hook can run for any operator under any design.
Cleanup has to happen somewhere else and later, which means the external job id
must have been persisted somewhere that survives the process.
Only durable operators have that. `ResumableJobMixin` writes the id to the
task state store. A
non-durable operator keeps it in instance attributes that die with the
process, so there is nothing
left to cancel with. Mode 2 is therefore fixable only for durable operators,
and it is fixable
precisely because the mixin already persists the id.
### Durable execution reduces this leak rather than causing it
Worth stating, because the problem surfaced during durable execution review
and reads like a
regression otherwise.
A non-durable operator orphans a job on every crashed attempt. Attempt 1
submits job A and dies,
attempt 2 submits B, attempt 3 submits C, and A and B are still running.
That is the duplicate
submission problem `ResumableJobMixin` exists to solve.
A durable operator reconnects on attempts 2 through N instead of
resubmitting, so it orphans once, at
the terminal attempt, rather than N times. This issue is about closing the
remaining case.
### Proposed direction
**For mode 1**, give operators a hook that fires when a task reaches a
terminal failed state, and have
`ResumableJobMixin` implement it by cancelling the tracked job. Operators
outside the mixin can adopt
it individually. `KubernetesPodOperator` shows the shape: cleanup on any
exit path, with an opt-out
for users who want the job to outlive the task, as
`on_finish_action="keep_pod"` allows today.
**For mode 2**, the worker is gone, so the options are server-side
reconciliation or having the next
attempt cancel the orphan it finds. Both need the persisted id, so both are
durable-only. This half is
larger and could be split into its own issue.
Open questions:
- New operator hook, or `ResumableJobMixin` wrapping `poll_until_complete`
in `try/except`?
- Should cleanup fire on every failed attempt or only the terminal one?
Cancelling on every attempt
would defeat the reconnect that makes retries work.
- What does the opt-out look like, and should cancelling be the default?
- Is mode 2 worth solving now, or is documenting it enough for a first pass?
--
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]
