This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 7aa410686c4 [v3-4-test] Load the serialized Dag in ti_run only for
stub tasks (#75795) (#75802)
7aa410686c4 is described below
commit 7aa410686c4d5fc3bf11ce6ad6ce012f487c39a5
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Oct 28 15:08:18 2026 +0800
[v3-4-test] Load the serialized Dag in ti_run only for stub tasks (#75795)
(#75802)
* [v3-4-test] Load the serialized Dag in ti_run only for stub tasks (#75795)
Starting any task deserialized its whole Dag so the route could find out
whether the task was a stub and read its arg_bindings. Denormalize the
serialized is_stub flag onto task_instance, which ti_run already fetches
and locks, and load the Dag only when it is set.
* Mark a Go or Java TriggerDagRun task a stub, as TypeScript does
* Inline the table name in the is_stub migration
* Starve a Dag the scheduler cannot deserialize
test_find_executable_task_instances_skips_dag_that_cannot_be_deserialized
fails on v3-4-test independently of this backport. #75745 brought the guard
over from #75263 without the starved set that #75263 relies on, because that
set came with #70307, which is not on this branch. The guard then returns
False without narrowing the next query, so a cycle whose batch is filled by
the bad Dag ends with nothing queued.
---------
---
airflow-core/docs/migrations-ref.rst | 4 +-
.../execution_api/routes/task_instances.py | 9 +++-
.../src/airflow/jobs/scheduler_job_runner.py | 7 ++-
.../0149_3_4_0_add_is_stub_to_task_instance.py | 54 +++++++++++++++++++++
airflow-core/src/airflow/models/taskinstance.py | 7 +++
airflow-core/src/airflow/utils/db.py | 2 +-
.../versions/head/test_task_instances.py | 45 ++++++++++++++++++
.../lang_sdk_fixtures/go_native.json | 2 +
.../lang_sdk_fixtures/java_native.json | 3 ++
.../tests/unit/dag_processing/test_dagbag.py | 9 ++++
.../tests/unit/models/test_taskinstance.py | 55 ++++++++++++++++++++++
go-sdk/airflow/serialize.go | 9 ++--
go-sdk/airflow/serialize_test.go | 3 +-
.../org/apache/airflow/sdk/execution/Serde.kt | 11 +++--
.../org/apache/airflow/sdk/TriggerDagRunTest.kt | 5 +-
ts-sdk/src/coordinator/serde.ts | 8 ++--
16 files changed, 213 insertions(+), 20 deletions(-)
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index 690ee0fc869..53476a7ac32 100644
--- a/airflow-core/docs/migrations-ref.rst
+++ b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description
|
+=========================+==================+===================+==============================================================+
-| ``ee86eed19e24`` (head) | ``9cffe90fa0c0`` | ``3.4.0`` | Add
pending_partition_key to asset_partition_dag_run and |
+| ``b4f1a7d2c903`` (head) | ``ee86eed19e24`` | ``3.4.0`` | Add is_stub
to task_instance. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``ee86eed19e24`` | ``9cffe90fa0c0`` | ``3.4.0`` | Add
pending_partition_key to asset_partition_dag_run and |
| | | | enforce
single pending row per key. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``9cffe90fa0c0`` | ``a5d7b9c13e40`` | ``3.4.0`` | Add
commit_allows_update to connection_test_request. |
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index d69e51cde62..bb3f8fdf20d 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -192,6 +192,7 @@ def ti_run(
TI.unixname,
TI.pid,
TI.dag_version_id,
+ TI.is_stub,
# This selects the raw JSON value, bypassing the deserialization
-- we want that to happen on the
# client
column("next_kwargs", JSON),
@@ -379,8 +380,12 @@ def ti_run(
# Only set for lang-SDK (foreign-runtime) tasks with a captured
TaskFlow arg
# spec; the route excludes unset fields, keeping regular responses
lean.
- if client_supports_arg_bindings() and (
- arg_bindings := get_arg_bindings(dag_bag, ti, session=session)
+ # ti.is_stub carries the row's own copy of the serialized flag and is
checked first, so the
+ # normal Python task path never loads the serialized Dag.
+ if (
+ ti.is_stub
+ and client_supports_arg_bindings()
+ and (arg_bindings := get_arg_bindings(dag_bag, ti,
session=session))
):
try:
context.arg_bindings =
get_arg_bindings_adapter().validate_python(arg_bindings)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 91304448aee..4597b2614da 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -583,6 +583,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
task_instance: TI,
concurrency_map: ConcurrencyMap,
session: Session,
+ starved_dags: set[str],
starved_tasks: set[tuple[str, str]],
starved_tasks_task_dagrun_concurrency: set[tuple[str, str, str]],
) -> bool:
@@ -601,8 +602,11 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
raise
except Exception:
# A single Dag that fails to deserialize must not stop the
scheduler from
- # scheduling other Dags; skip this task instance instead of
crashing the loop.
+ # scheduling other Dags; skip this task instance instead of
crashing the loop. Starving
+ # the Dag keeps the next query of this cycle from handing back the
same task instance,
+ # which would end the cycle with nothing queued whenever a bad Dag
fills the batch.
self.log.exception("Failed to load Dag %s for run %s, skipping
it", dag_id, run_id)
+ starved_dags.add(dag_id)
return False
# If the DAG is missing, fail all scheduled TIs for this DAG.
@@ -921,6 +925,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
task_instance=task_instance,
concurrency_map=concurrency_map,
session=session,
+ starved_dags=starved_dags,
starved_tasks=starved_tasks,
starved_tasks_task_dagrun_concurrency=(starved_tasks_task_dagrun_concurrency),
)
diff --git
a/airflow-core/src/airflow/migrations/versions/0149_3_4_0_add_is_stub_to_task_instance.py
b/airflow-core/src/airflow/migrations/versions/0149_3_4_0_add_is_stub_to_task_instance.py
new file mode 100644
index 00000000000..48cc4c5870e
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0149_3_4_0_add_is_stub_to_task_instance.py
@@ -0,0 +1,54 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add is_stub to task_instance.
+
+Revision ID: b4f1a7d2c903
+Revises: ee86eed19e24
+Create Date: 2026-10-27 09:12:44.317602
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.migrations.utils import sqlite_rebuilds
+
+# revision identifiers, used by Alembic.
+revision = "b4f1a7d2c903"
+down_revision = "ee86eed19e24"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+ """Add is_stub to task_instance."""
+ with op.batch_alter_table("task_instance") as batch_op:
+ batch_op.add_column(sa.Column("is_stub", sa.Boolean(), nullable=True))
+
+
+def downgrade():
+ """Remove is_stub from task_instance."""
+ # Dropping a column rebuilds the table on SQLite, unlike adding one, so
the foreign keys of
+ # everything that points at task_instance have to be held off while the
rows are copied.
+ with sqlite_rebuilds(op), op.batch_alter_table("task_instance") as
batch_op:
+ batch_op.drop_column("is_stub")
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index ca06135a3f5..6905ae965d8 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/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
+ # to a new Dag version without being refreshed from its task keeps the
value it was created with,
+ # which fails closed: a task that has become a stub loses its arg_bindings
rather than running wrong.
+ is_stub: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
queued_dttm: Mapped[datetime | None] = mapped_column(UtcDateTime,
nullable=True)
scheduled_dttm: Mapped[datetime | None] = mapped_column(UtcDateTime,
nullable=True)
queued_by_job_id: Mapped[int | None] = mapped_column(Integer,
nullable=True)
@@ -1385,6 +1390,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
"executor_config": task.executor_config,
"operator": task.task_type,
"custom_operator_name": getattr(task, "operator_name", None),
+ "is_stub": task.is_stub,
"region_index": region_index,
"region_id": region_id,
"_task_display_property_value": task.task_display_name,
@@ -1581,6 +1587,7 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
self.operator = task.task_type
op_name = getattr(task, "operator_name", None)
self.custom_operator_name = op_name if isinstance(op_name, str) else ""
+ self.is_stub = task.is_stub
# Re-apply cluster policy here so that task default do not overload
previous data
task_instance_mutation_hook(self, dag_run=dag_run)
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index b4a1638422f..61e80cbb6a6 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -117,7 +117,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.1.8": "509b94a1042d",
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
- "3.4.0": "ee86eed19e24",
+ "3.4.0": "b4f1a7d2c903",
}
# Prefix used to identify tables holding data moved during migration.
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 391d55aeed8..9ff1bb8433d 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -1081,6 +1081,49 @@ class TestTIRunState:
assert response.status_code == 200
assert "arg_bindings" not in response.json()
+ def test_ti_run_does_not_load_the_dag_for_a_regular_task(self, client,
dag_maker, mocker):
+ """The row rules the task out, so starting it reads no serialized Dag
-- the point of the column."""
+ with dag_maker("test_regular_task_arg_bindings", serialized=True):
+
+ @task
+ def transform(country: str): ...
+
+ transform("uk")
+
+ dr = dag_maker.create_dagrun()
+ (ti,) = dr.get_task_instances()
+ assert ti.is_stub is False
+ ti.set_state(State.QUEUED)
+ dag_maker.session.flush()
+ get_dag = mocker.patch.object(DBDagBag, "get_dag", autospec=True)
+
+ response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=self.RUN_PAYLOAD)
+
+ assert response.status_code == 200
+ assert "arg_bindings" not in response.json()
+ get_dag.assert_not_called()
+
+ def test_ti_run_treats_an_unset_is_stub_as_a_regular_task(self, client,
dag_maker):
+ """A row written before the column existed carries null and keeps its
pre-upgrade behavior."""
+ with dag_maker("test_unset_is_stub", serialized=True):
+
+ @task.stub
+ def transform(country: str): ...
+
+ transform("uk")
+
+ dr = dag_maker.create_dagrun()
+ (ti,) = dr.get_task_instances()
+ ti.set_state(State.QUEUED)
+ dag_maker.session.get(TaskInstance, ti.id).is_stub = None
+ dag_maker.session.commit()
+ assert
dag_maker.session.scalar(select(TaskInstance.is_stub).where(TaskInstance.id ==
ti.id)) is None
+
+ response = client.patch(f"/execution/task-instances/{ti.id}/run",
json=self.RUN_PAYLOAD)
+
+ assert response.status_code == 200
+ assert "arg_bindings" not in response.json()
+
@mock.patch(
"airflow.api_fastapi.execution_api.routes.task_instances.get_arg_bindings",
autospec=True,
@@ -2033,6 +2076,8 @@ class TestTIRunState:
dag_id=str(uuid4()),
)
ti.queued_dttm = queued_at
+ # The gate short-circuits without this, so the mocked spec below never
reaches the route.
+ ti.is_stub = True
session.commit()
with mock.patch(
diff --git
a/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json
b/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json
index 031f73bebd1..8b440945e3b 100644
--- a/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json
+++ b/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/go_native.json
@@ -1621,6 +1621,7 @@
"ratio": 0.5,
"rows": 2
},
+ "is_stub": true,
"logical_date": "2026-09-30 00:00:00+00:00",
"skip_when_already_exists": true,
"task_id": "trigger_all",
@@ -1649,6 +1650,7 @@
"Triggered DAG": "_link_TriggerDagRunLink"
},
"_task_module":
"airflow.providers.standard.operators.trigger_dagrun",
+ "is_stub": true,
"logical_date": "NOTSET",
"skip_when_already_exists": false,
"task_id": "trigger_default",
diff --git
a/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/java_native.json
b/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/java_native.json
index 50314784b10..53f0a7256d7 100644
--- a/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/java_native.json
+++ b/airflow-core/tests/unit/dag_processing/lang_sdk_fixtures/java_native.json
@@ -27,6 +27,7 @@
"task_type": "DumpExtract",
"_task_module": "org.apache.airflow.sdk.execution",
"language": "java",
+ "is_stub": true,
"template_fields": [],
"retries": 2,
"queue": "java",
@@ -44,6 +45,7 @@
"task_type": "DumpLoad",
"_task_module": "org.apache.airflow.sdk.execution",
"language": "java",
+ "is_stub": true,
"template_fields": [],
"queue": "java",
"trigger_rule": "all_done"
@@ -112,6 +114,7 @@
"task_type": "DumpExtract",
"_task_module": "org.apache.airflow.sdk.execution",
"language": "java",
+ "is_stub": true,
"template_fields": []
}
}
diff --git a/airflow-core/tests/unit/dag_processing/test_dagbag.py
b/airflow-core/tests/unit/dag_processing/test_dagbag.py
index d24443f709f..7bcd745c24c 100644
--- a/airflow-core/tests/unit/dag_processing/test_dagbag.py
+++ b/airflow-core/tests/unit/dag_processing/test_dagbag.py
@@ -1660,6 +1660,15 @@ 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, and a language
runtime runs every task of a
+ Dag defined in it, a TriggerDagRun one included, so each has to say so."""
+ dag = DagSerialization.from_dict(data)
+ for task_id, task in dag.task_dict.items():
+ assert task.is_stub, task_id
+
+
def _serialize_native_dag(dag_id: str, fileloc: str, relative_fileloc: str) ->
dict:
with DAG(dag_id, schedule=None) as dag:
BaseOperator(task_id="extract")
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py
b/airflow-core/tests/unit/models/test_taskinstance.py
index 286a04574db..958ed9c8252 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -2865,6 +2865,7 @@ class TestTaskInstance:
"priority_weight": 123,
"operator": "some_custom_operator",
"custom_operator_name": "some_custom_operator",
+ "is_stub": False,
"queued_dttm": run_date + datetime.timedelta(hours=1),
"scheduled_dttm": run_date + datetime.timedelta(hours=1),
"rendered_map_index": None,
@@ -5069,6 +5070,60 @@ class TestMakeTaskCarrier:
assert "traceparent" in carrier
[email protected]("is_stub", [True, False])
+def test_refresh_from_task_copies_is_stub(is_stub):
+ """ti_run gates on the row, so the flag has to travel from the task to the
TaskInstance."""
+ ser_task = create_scheduler_operator(EmptyOperator(task_id="empty"))
+ ser_task.is_stub = is_stub
+
+ ti = TI(ser_task, run_id=None, dag_version_id=mock.MagicMock())
+
+ assert ti.is_stub is is_stub
+
+
[email protected]_test
[email protected](
+ ("decorator", "expected"),
+ [pytest.param(task.stub, True, id="stub"), pytest.param(task, False,
id="regular")],
+)
+def test_scheduled_task_instances_carry_is_stub(dag_maker, decorator,
expected):
+ """A scheduled row carries the flag, whichever of the two write paths
filled it in."""
+ with dag_maker("test_scheduled_is_stub", serialized=True):
+
+ @decorator
+ def transform(country: str): ...
+
+ transform("uk")
+
+ (ti,) = dag_maker.create_dagrun().get_task_instances()
+
+ assert ti.is_stub is expected
+
+
[email protected]_test
[email protected]("is_stub", [True, False])
+def test_insert_mapping_carries_is_stub(dag_maker, session, is_stub):
+ """The scheduler bulk-inserts rows from this mapping, so the flag has to
be in it."""
+ with dag_maker("test_insert_mapping_is_stub"):
+ EmptyOperator(task_id="t1")
+ session.flush()
+
+ op = create_scheduler_operator(dag_maker.dag.get_task("t1"))
+ op.is_stub = is_stub
+ dag_run = mock.MagicMock()
+ dag_run.context_carrier = new_dagrun_trace_carrier()
+
+ mapping = TaskInstance.insert_mapping(
+ run_id="test_run",
+ task=op,
+ map_index=0,
+ dag_version_id=None,
+ dag_run=dag_run,
+ )
+
+ assert mapping["is_stub"] is is_stub
+
+
@pytest.mark.db_test
def test_insert_mapping_includes_context_carrier(dag_maker, session):
"""insert_mapping should include a context_carrier with a traceparent
derived from the dag run."""
diff --git a/go-sdk/airflow/serialize.go b/go-sdk/airflow/serialize.go
index bd294b85b4f..0eec1a44a3b 100644
--- a/go-sdk/airflow/serialize.go
+++ b/go-sdk/airflow/serialize.go
@@ -424,9 +424,6 @@ func (d *DagRef) serializeTaskLocked(task *TaskRef)
map[string]any {
// Python writes template_fields for every operator. A Go task
has no template fields, so the list
// is empty.
data["template_fields"] = []string{}
- // is_stub makes the API server send the task its arguments
from _arg_bindings when the
- // task runs, as it does for a @task.stub task of a Python Dag.
- data["is_stub"] = true
if len(task.inputs) > 0 {
data["_arg_bindings"] =
serializeArgBindings(task.inputs)
}
@@ -437,7 +434,11 @@ func (d *DagRef) serializeTaskLocked(task *TaskRef)
map[string]any {
data["_can_skip_downstream"] = true
}
}
- // A task from TriggerDagRun takes the queue of the Dag too, because
the Go runtime runs it.
+ // The Go runtime runs every task of a Go Dag, a task from
TriggerDagRun included, so every one of
+ // them is a stub. is_stub makes the API server send the task its
arguments from _arg_bindings when
+ // the task runs, as it does for a @task.stub task of a Python Dag.
+ data["is_stub"] = true
+ // The worker picks a runtime by queue, so a task from TriggerDagRun
takes the queue of the Dag too.
if spec.Queue == "" {
spec.Queue = d.spec.Queue
}
diff --git a/go-sdk/airflow/serialize_test.go b/go-sdk/airflow/serialize_test.go
index 4f728ada5d8..4a7bcf96dd8 100644
--- a/go-sdk/airflow/serialize_test.go
+++ b/go-sdk/airflow/serialize_test.go
@@ -925,6 +925,7 @@ func
TestSerializeWritesATriggerDagRunAsATriggerDagRunOperator(t *testing.T) {
assertJSON(t, `{
"task_id": "trigger",
"task_type": "TriggerDagRunOperator",
+ "is_stub": true,
"_task_module":
"airflow.providers.standard.operators.trigger_dagrun",
"ui_color": "#ffefeb",
"template_fields": [
@@ -1012,7 +1013,7 @@ func
TestSerializeWritesOnlyTheTriggerDagRunOptionsThatAreSet(t *testing.T) {
"skip_when_already_exists": false
}`, withoutKeys(serializedTask(t, got, "trigger"),
"task_id", "task_type", "_task_module", "ui_color",
"template_fields",
- "template_fields_renderers", "_operator_extra_links",
+ "template_fields_renderers", "_operator_extra_links", "is_stub",
))
// An empty FailedStates means that no Dag run state fails the task.
assertJSON(t, `[]`, serializedTask(t, got,
"trigger_never_fails")["failed_states"])
diff --git
a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
index c01592b8eea..8a5d470426a 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
@@ -145,15 +145,16 @@ private fun serializeTask(
// value never matches the tuple default it is compared against), so it
// is unconditional here too. Java tasks have no template fields.
data["template_fields"] = emptyList<Any?>()
- // What marks a task whose arguments Airflow resolves per instance for a
- // runtime outside Python, as `@task.stub` does on the Python side.
- // `get_arg_bindings` reads nothing without it.
- data["is_stub"] = true
argBindings(taskId, def)?.let { data["_arg_bindings"] = it }
}
+ // The Java runtime runs every task of a Java Dag, a trigger task included,
so
+ // every one of them is a stub. is_stub marks a task whose arguments Airflow
+ // resolves per instance for a runtime outside Python, as `@task.stub` does
on
+ // the Python side. `get_arg_bindings` reads nothing without it.
+ data["is_stub"] = true
// Lets NotPreviouslySkippedDep re-skip a cleared downstream, as Python's
SkipMixin does.
if (def.decider != null) data["_can_skip_downstream"] = true
- // A trigger task takes the Dag's queue too, because the Java runtime runs
it.
+ // The worker picks a runtime by queue, so a trigger task takes the Dag's
queue too.
val config =
if (dagQueue == null || "queue" in def.configValues) def.configValues else
def.configValues + ("queue" to dagQueue)
// Emit only config entries that differ from their schema default, mirroring
diff --git
a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TriggerDagRunTest.kt
b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TriggerDagRunTest.kt
index 56c0bb410b7..b7bef8724f7 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TriggerDagRunTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/TriggerDagRunTest.kt
@@ -168,9 +168,10 @@ internal class TriggerDagRunTest {
assertEquals("NOTSET", data["logical_date"])
assertEquals(false, data["wait_for_completion"])
assertEquals(false, data["skip_when_already_exists"])
- // A Java task's markers belong to a task the Java runtime runs a body for.
+ // The Java runtime runs a trigger task too, so it carries the stub marker.
+ assertEquals(true, data["is_stub"])
+ // `language` marks a task the Java runtime runs a body for, which this is
not.
assertFalse("language" in data)
- assertFalse("is_stub" in data)
assertFalse("reset_dag_run" in data)
}
diff --git a/ts-sdk/src/coordinator/serde.ts b/ts-sdk/src/coordinator/serde.ts
index c528980487c..b6938fa81d6 100644
--- a/ts-sdk/src/coordinator/serde.ts
+++ b/ts-sdk/src/coordinator/serde.ts
@@ -252,9 +252,11 @@ function serializeTask(
// matches the tuple default it is compared against. A TypeScript task has
// no Jinja templating, so the list is empty rather than absent.
template_fields: [],
- // What marks a task whose arguments the API server resolves per instance
- // and sends to a foreign runtime, as `@task.stub` does on the Python side.
- // `get_arg_bindings` reads nothing without it.
+ // The TypeScript runtime runs every task of a TypeScript Dag, a trigger
task
+ // included, so every one of them is a stub. `is_stub` marks a task whose
+ // arguments the API server resolves per instance and sends to a foreign
+ // runtime, as `@task.stub` does on the Python side. `get_arg_bindings`
reads
+ // nothing without it.
is_stub: true,
};
const label = `task "${taskId}" of Dag "${dagId}"`;
