1fanwang opened a new pull request, #70813:
URL: https://github.com/apache/airflow/pull/70813
Retry policies are documented as running in the task worker process, but the
consequence of that is easy to miss. The policy is evaluated from a caught
exception, so a failure where the worker process never raises one is not
covered: an external kill, a node drain, a spot-instance reclaim, the OOM
killer. Those attempts fall through to the standard `retries` count.
The design is reasonable and the docs already say *where* the policy runs.
What they do not say is what happens when there is no exception to hand it. On
preemptible or spot capacity that is the common failure mode rather than an
edge case, and a policy written as `RetryRule(exception=...,
action=RetryAction.FAIL)` looks like it is simply being ignored.
This records the boundary in two places: a row in the "Composition with
existing parameters" table next to the existing `AirflowFailException` note,
and the `RetryPolicy` docstring. Docs only, no behaviour change.
##### Testing
Same `ExceptionRetryPolicy` on two tasks, both `retries=3`. The policy fails
everything immediately, so if it is consulted the task must end on try 1. Run
on 3.4.0 against a real deployment (api-server + scheduler + LocalExecutor,
task in a supervised subprocess), not a mocked runner.
| Task | How it failed | Policy consulted | Attempts used | Final state |
|---|---|---|---|---|
| `policy_raises` | `raise ConnectionError(...)` | yes | 1 of 4 | `failed` |
| `policy_worker_killed` | `os.kill(os.getpid(), SIGKILL)` | no | 4 of 4 |
`failed` |
Identical policy, opposite outcome, decided entirely by how the worker died.
<details>
<summary>Raw logs</summary>
Task body used for the termination case:
```python
@task(retries=3, retry_delay=timedelta(seconds=1),
retry_policy=FAIL_EVERYTHING)
def killed():
os.kill(os.getpid(), signal.SIGKILL)
```
Exception case, policy consulted and honoured:
```
[info] Retry policy decision [task] action=fail loc=task_runner.py:1795
reason='POLICY WAS CONSULTED'
[info] Task instance state updated ... new_state=failed rows_affected=1
```
Termination case, policy never consulted across all four attempts:
```
$ for f in
logs/dag_id=policy_worker_killed/run_id=kill-e2e-1/task_id=killed/attempt=*.log;
do
echo "$(basename $f): $(grep -c 'Retry policy decision' $f)"
done
attempt=1.log: 0
attempt=2.log: 0
attempt=3.log: 0
attempt=4.log: 0
```
The first attempt log is zero bytes, since the process died before flushing:
```
-rw-r--r-- 1 0 attempt=1.log
-rw-r--r-- 1 2325 attempt=2.log
-rw-r--r-- 1 2325 attempt=3.log
-rw-r--r-- 1 2325 attempt=4.log
```
Scheduler side, the retries are spent and the policy never gets a say:
```
[info] Marking task as UP_FOR_RETRY. dag_id=policy_worker_killed,
task_id=killed, run_id=kill-e2e-1
[info] Marking task as UP_FOR_RETRY. dag_id=policy_worker_killed,
task_id=killed, run_id=kill-e2e-1
[info] TaskInstance Finished: dag_id=policy_worker_killed, task_id=killed,
state=failed,
executor_state=success, try_number=4, max_tries=3,
operator=_PythonDecoratedOperator
```
</details>
---
##### Was generative AI tooling used to co-author this PR?
- [ ] Yes (please specify the tool below)
--
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]
