kaxil commented on code in PR #75795:
URL: https://github.com/apache/airflow/pull/75795#discussion_r4390962250
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -382,7 +383,10 @@ def ti_run(
if first_task_reschedule_start_date is not None:
context.first_task_reschedule_start_date =
first_task_reschedule_start_date
- serialized_task = get_task(dag_bag, ti, session=session)
+ # Avoid loading the serialized Dag on the normal Python task path:
only a lang-SDK
+ # (foreign-runtime) task needs its definition here, and ti.is_stub
carries the row's own copy
+ # of the serialized flag.
+ serialized_task = get_task(dag_bag, ti, session=session) if ti.is_stub
else None
Review Comment:
Not for this PR, since the description already scopes the `main` timeout
path out, but a heads-up for that redesign: Go and Java `TriggerDagRun` tasks
also fall through this check. Neither serializer writes `is_stub` for them
(`writeTriggerDagRun` in `go-sdk/airflow/serialize.go`, `serializeTask` in the
Java `Serde.kt`), yet they take the Dag's queue and the language runtime runs
them. That coordinator arms the timeout only from
`ti_context.execution_timeout` (`coordinators/_subprocess.py`), so on `main` a
Go trigger task with `ExecutionTimeout` and `WaitForCompletion` loses its
timeout, and #75387 only covers the Python runner. Having the Go and Java
serializers write `is_stub` on every task, the way
`ts-sdk/src/coordinator/serde.ts` does, would close it.
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -1141,6 +1141,11 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
priority_weight: Mapped[int | None] = mapped_column(Integer, nullable=True)
operator: Mapped[str | None] = mapped_column(String(1000), nullable=True)
custom_operator_name: Mapped[str | None] = mapped_column(String(1000),
nullable=True)
+ # Denormalized from the task so ti_run can tell a stub (foreign-runtime)
task from a regular one
+ # without loading the serialized Dag. Null on rows written before the
column existed. A row moved
Review Comment:
3.4.0b1 and b2 already send `arg_bindings` from `ti_run` but have no
`is_stub` column, so a stub TI queued before upgrading from a beta reads null
and runs without its arguments. A retry carries the null forward too, since
`prepare_db_for_next_try` copies the column. Would `if ti.is_stub is not False`
work in the route? Rows written after the migration always have a bool, so it
only changes what happens to these old rows.
##########
airflow-core/tests/unit/dag_processing/test_dagbag.py:
##########
@@ -1660,6 +1660,17 @@ def test_recorded_runtime_payloads_are_valid(data):
DagSerialization.validate_serialized_dag(data)
[email protected]("data", _recorded_payloads())
+def test_recorded_runtime_payloads_mark_their_tasks_as_stubs(data):
+ """ti_run derives arg_bindings only for a stub task, so a Dag defined in a
language SDK has to say so."""
+ dag = DagSerialization.from_dict(data)
+ for task_id, task in dag.task_dict.items():
+ # A Go or Java TriggerDagRun task is the Python operator itself and
runs in-process. The
Review Comment:
This contradicts the Go SDK: the `TriggerDagRun` doc says the Go runtime
runs the task and the Dag needs no Python worker for it, and both the Go and
Java serializers put it on the Dag's queue for that reason. Whatever happens
with the route comment, I don't think the test should pin these as non-stubs on
the basis that they run in Python.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -2108,6 +2153,8 @@ def
test_ti_run_skips_queued_duration_metric_when_the_response_fails(
dag_id=str(uuid4()),
)
ti.queued_dttm = queued_at
+ # The route reads a serialized spec only for a stub task, which is
what fails below.
+ ti.is_stub = True
Review Comment:
`get_arg_bindings` is patched with a fixed `return_value`, so the 500
happens whether or not the route loads the task, and the test passes without
these two lines. I'd drop them, since the comment points at the gate as the
thing that fails.
--
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]
