dabla commented on code in PR #75774:
URL: https://github.com/apache/airflow/pull/75774#discussion_r4390371361
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/sensors/msgraph.py:
##########
@@ -119,26 +134,30 @@ def execute(self, context: Context):
headers=self.headers,
data=self.data,
conn_id=self.conn_id,
- timeout=self.timeout,
proxies=self.proxies,
scopes=self.scopes,
api_version=self.api_version,
serializer=type(self.serializer),
),
method_name=self.execute_complete.__name__,
+ kwargs={"sensor_started_at": started_at},
+ timeout=timedelta(seconds=remaining) if remaining is not None else
None,
)
- def retry_execute(
- self,
- context: Context,
- **kwargs,
- ) -> Any:
- self.execute(context=context)
+ @staticmethod
+ def _anchor(sensor_started_at: float | None) -> float:
Review Comment:
**[nit]** One concept, three names: `_anchor`, `started_at`,
`sensor_started_at` (and "anchor" again in the test names).
They all mean the moment the try started. Picking one (`started_at` reads
best next to `BaseSensorOperator.execute`, which uses the same word) and naming
the method after what it returns, e.g. `_started_at(sensor_started_at)`, keeps
the code and the tests in one vocabulary ([Fowler,
*UbiquitousLanguage*](https://martinfowler.com/bliki/UbiquitousLanguage.html)).
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/sensors/msgraph.py:
##########
@@ -183,11 +202,22 @@ def execute_complete(
return result
- # Re-defer with timeout so Airflow enforces the sensor timeout
natively
+ started_at = self._anchor(sensor_started_at)
+ remaining = self._remaining_timeout(started_at)
+ if remaining is not None and remaining <= 0:
+ raise AirflowSensorTimeout(
+ f"Sensor has timed out; run duration of {self.timeout
- remaining} seconds "
+ f"exceeds the specified timeout of {self.timeout}."
+ )
+ delay = (
+ self.retry_delay
+ if remaining is None
+ else min(self.retry_delay, timedelta(seconds=remaining))
+ )
self.defer(
- trigger=TimeDeltaTrigger(self.retry_delay),
+ trigger=TimeDeltaTrigger(delay),
Review Comment:
**[warning]** The sleep defer carries no `timeout`, so the deadline is
cleared for the whole sleep.
A defer without `timeout` stores `trigger_timeout=None` (execution API
`ti_update_state`, and `TaskInstance._defer_task` on 2.x), which is the same
clearing this PR removes from the request defer. While the task sleeps on the
`TimeDeltaTrigger`, core has no deadline for it: a trigger that never fires
(triggerer stuck, trigger lost) leaves the task deferred until
`execution_timeout`, if any. #62241 had put a timeout on this defer; the fix
here drops it instead of making it stop sliding.
Suggest keeping a timeout on the sleep defer that lies above the sleep
itself so the "one last poll" path is never raced by it, for example:
```python
self.defer(
trigger=TimeDeltaTrigger(delay),
method_name=self.retry_execute.__name__,
kwargs={"sensor_started_at": started_at},
timeout=None if remaining is None else timedelta(seconds=remaining) +
self.retry_delay,
)
```
A lost sleep trigger then ends through `TaskDeferralTimeout`, which
`BaseSensorOperator.resume_execution` turns into `AirflowSensorTimeout` with
`soft_fail`/`never_fail` applied, the same path the request defer already
relies on. One test for the sleep defer's timeout next to
`test_execute_complete_redefers_with_capped_sleep_and_anchor` would pin it.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/sensors/msgraph.py:
##########
@@ -119,26 +134,30 @@ def execute(self, context: Context):
headers=self.headers,
data=self.data,
conn_id=self.conn_id,
- timeout=self.timeout,
proxies=self.proxies,
scopes=self.scopes,
api_version=self.api_version,
serializer=type(self.serializer),
),
method_name=self.execute_complete.__name__,
+ kwargs={"sensor_started_at": started_at},
+ timeout=timedelta(seconds=remaining) if remaining is not None else
None,
)
- def retry_execute(
- self,
- context: Context,
- **kwargs,
- ) -> Any:
- self.execute(context=context)
+ @staticmethod
+ def _anchor(sensor_started_at: float | None) -> float:
+ return sensor_started_at if sensor_started_at is not None else
timezone.utcnow().timestamp()
+
+ def _remaining_timeout(self, started_at: float) -> float | None:
+ if self.timeout is None:
Review Comment:
**[nit]** `self.timeout is None` cannot happen; the three `None` branches
and `test_timeout_attribute_none_never_times_out` cover a state the base class
rejects.
`BaseSensorOperator.__init__` coerces `timeout` to a number
(`self._coerce_timeout(timeout).total_seconds()`) and raises on anything else,
on the Task SDK and on 2.11 alike, with a 7-day default. The test only reaches
the branch by assigning `sensor.timeout = None` after construction. Typing
`_remaining_timeout` as `-> float`, dropping the `None` checks in
`retry_execute`, `_defer_request` and `execute_complete`, and removing that
test leaves less to read for the same behaviour ([Fowler,
*Yagni*](https://martinfowler.com/bliki/Yagni.html)).
--
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]
