vishalmore90 opened a new issue, #73475:
URL: https://github.com/apache/airflow/issues/73475
### Under which category would you file this issue?
Task SDK
### Apache Airflow version
main (3.0.0.dev0)
### What happened and how to reproduce it?
In the new Airflow 3 Task SDK (`task_runner.py`), when a task defines an
`execution_timeout` and subsequently enters a deferred state, the execution
timeout logic resets completely upon resumption.
**Issue Description**
In `task-sdk/src/airflow/sdk/execution_time/task_runner.py`,
`_run_execute_callable` applies the `execution_timeout` using a simple
`timeout_seconds = task.execution_timeout.total_seconds()`:
```python
if task.execution_timeout:
from airflow.sdk.execution_time.timeout import timeout
# TODO: handle timeout in case of deferral
timeout_seconds = task.execution_timeout.total_seconds()
try:
# It's possible we're already timed out, so fast-fail if true
if timeout_seconds <= 0:
raise AirflowTaskTimeout()
# Run task in timeout wrapper
with timeout(timeout_seconds):
result = ctx.run(execute, context=context)
```
Because `task.execution_timeout.total_seconds()` evaluates to the static
configured timeout (e.g. `timedelta(minutes=60)`), any time previously spent
executing or waiting in a deferred state is ignored. As a result, the task gets
an entirely new 60-minute execution window every single time it resumes from
deferral.
**Steps to reproduce:**
1. Create a DAG with a task that uses a deferrable operator (e.g.
`TimeDeltaSensorAsync` or a custom deferrable operator) and set
`execution_timeout=timedelta(minutes=5)`.
2. Configure the task to run for 1 minute, defer for 3 minutes, and then
resume.
3. Upon resumption, the task will erroneously be granted a fresh 5-minute
timeout window instead of the remaining 1 minute, effectively allowing the task
to run up to 9 total minutes before timing out (or indefinitely if it defers
repeatedly).
### What you think should happen instead?
The execution timeout should cover the entire runtime of the task, including
time already spent prior to and during the deferral (maintaining parity with
Airflow 2.x behavior).
When determining `timeout_seconds` inside `_run_execute_callable` (or before
it), the `task_runner` should compute the remaining time by subtracting the
already elapsed time since `ti.start_date`. For example:
```python
timeout_seconds = task.execution_timeout.total_seconds()
# Calculate time already spent if resuming
if getattr(ti, "start_date", None):
from airflow.utils import timezone
elapsed = (timezone.utcnow() - ti.start_date).total_seconds()
timeout_seconds -= elapsed
```
*(Note: `ti` context or elapsed duration would need to be made available to
`_run_execute_callable`)*
### Operating System
_No response_
### Deployment
Other
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
_No response_
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
_No response_
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
