henry3260 commented on code in PR #74869:
URL: https://github.com/apache/airflow/pull/74869#discussion_r4302690565
##########
airflow-core/src/airflow/api_fastapi/execution_api/services/task_instances.py:
##########
@@ -44,18 +45,42 @@ def client_supports_arg_bindings() -> bool:
return AddArgBindingsToTIRunContext.is_applied
-def get_arg_bindings(dag_bag: DBDagBag, ti: Any, *, session: Session) -> list
| None:
+def get_stub_task(dag_bag: DBDagBag, ti: Any, *, session: Session) ->
SerializedOperator | None:
"""
- Extract the stub task's TaskFlow arg spec from its Dag version.
+ Return the task of *ti* from its Dag version if a language SDK runs it,
else ``None``.
- Mapped (``.expand()``) stubs never capture a parse-time spec, so they
resolve to
- ``None`` here and keep the legacy ignored-args behavior; per-map-index
delivery
- lands in a follow-up.
+ Both ``@task.stub`` tasks and the tasks of a Dag defined natively in a
language SDK
+ serialize with ``is_stub=True``.
"""
if ti.dag_version_id is None:
return None
if (dag := dag_bag.get_dag(ti.dag_version_id, session=session)) is None:
return None
if (task := dag.task_dict.get(ti.task_id)) is None or not task.is_stub:
return None
+ return task
+
+
+def get_arg_bindings(dag_bag: DBDagBag, ti: Any, *, session: Session) -> list
| None:
+ """
+ Extract the stub task's TaskFlow arg spec from its Dag version.
+
+ Mapped (``.expand()``) stubs never capture a parse-time spec, so they
resolve to
+ ``None`` here and keep the legacy ignored-args behavior; per-map-index
delivery
+ lands in a follow-up.
+ """
+ if (task := get_stub_task(dag_bag, ti, session=session)) is None:
+ return None
return task.arg_bindings
+
+
+def get_language_sdk_execution_timeout(dag_bag: DBDagBag, ti: Any, *, session:
Session) -> float | None:
Review Comment:
`get_language_sdk_execution_timeout` uses a falsy check, so a negative
`execution_timeout` passes through: `timedelta(seconds=-1)` is truthy,
returns
`-1.0`, and `_arm_execution_timeout` lands the deadline at
`monotonic() - 1 + KILLED_TASK_CLEANUP_TIME` SIGTERM 59s later rather than at
once. `bases/operator.py` only type-checks the field, so this is reachable.
The Python task runner fast-fails instead (`if timeout_seconds <= 0: raise
AirflowTaskTimeout()`, right before it sends `SetExecutionTimeout`), which
is why
the supervisor never sees a non-positive value today.
- Treat non-positive as no timeout, which keeps the fix inside the helper and
matches what the docstring already says about zero:
```python
if (task := get_stub_task(dag_bag, ti, session=session)) is None:
return None
timeout_seconds = task.execution_timeout.total_seconds() if
task.execution_timeout else 0
if timeout_seconds <= 0:
return None
return timeout_seconds
--
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]
