Fix AwaitMessageTriggerFunctionSensor not honoring timeout

[TimurRakhmatullin86]

AwaitMessageTriggerFunctionSensor accepts a timeout parameter and its docstring promises "Time elapsed before the task times out and fails", but neither defer() call passed timeout=, so the trigger could wait indefinitely and the parameter was silently ignored.

This applies the same fix that was merged for the sibling AwaitMessageSensor in #62104 (issue #62097): convert the numeric sensor timeout to a timedelta and pass it to defer() — here in both execute and execute_complete, since this sensor re-defers after every processed event.

Semantics note: because this sensor re-defers indefinitely by design, the timeout applies to each await cycle rather than to total task runtime — i.e. the task fails (or skips, with soft_fail=True) if no matching message arrives within timeout seconds of the most recent deferral. Total-elapsed semantics would unconditionally kill a healthy, indefinitely-running sensor, so per-cycle is the useful interpretation; happy to adjust if maintainers prefer otherwise.

On deferral timeout the standard sensor machinery converts TaskDeferralTimeout into AirflowSensorTimeout, and soft_fail is honored.

Adds a regression test asserting the timeout is forwarded on both deferral paths (it was None before this fix).

[potiuk]

Thanks — correct diagnosis and a well-targeted fix.

self.defer(...) was called without timeout= in both execute() and execute_complete(), so AwaitMessageTriggerFunctionSensor accepted a timeout and then never honoured it — the deferral had no deadline and the task could wait forever. The conversion to timedelta is necessary too, since BaseOperator.defer() expects one while BaseSensorOperator.timeout is float seconds.

The test is the right one. Note that the existing test_await_message_trigger_function_with_timeout_parameter only asserted sensor.timeout == 600 — that the value was stored, not that it reached the deferral — so it would have stayed green throughout the bug. Asserting exc_info.value.timeout == timedelta(seconds=600) is what actually pins the behaviour, and covering execute_complete as well as execute matters here because this sensor re-defers after every processed event; a fix that only handled the first deferral would leave every subsequent one unbounded.

One tidy-up worth doing whenever you next touch this file, not blocking: the four-line int/float->timedelta conversion is duplicated verbatim in both methods. A small _deferral_timeout property (or module-level helper) would keep the two deferrals from drifting apart later — which is precisely how one of them ended up missing timeout= in the first place.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting