You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ja...@apache.org on 2026/09/29 13:18:16 UTC
(airflow) branch jason/lang-sdk-e2e/08-native-e2e updated (92d2b908f5e -> dedaa3048e3)
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a change to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
omit 92d2b908f5e Build the k8s TypeScript artifacts only for the native Dag test
omit 480049ba3c3 Share the Dag run lookup and the native Java queue name
omit a2b2ec9ba1d Run the Lang-SDK e2e jobs when native Dag parsing changes
omit 1ab065690f1 Add a native Java Dag to the compose e2e
omit 4011695dbfb Keep the containerized Java build's Maven repository on the host
omit 79328eddc5a Check the native TypeScript Dag's source and triggered run
omit 72f3c3fcbdb Parse the native TypeScript Dag in the compose e2e
omit 7f6dfc790fe TS SDK: cover native TypeScript Dags end to end
omit 39c93a66bea Java SDK: Serialize native Dags to DagSerialization v3 on parse requests
omit 58ea38ce7db Java SDK: Declare a Dag's task graph with @Builder.Deps
omit b2b28cf9c84 Java SDK: Resolve a task's arguments from the Dag's own wiring
omit c3651ae98d4 Java SDK: Generate Dag and task configuration from the serialization schema
omit 9cbc0740e67 Java SDK: Describe native Java Dags without reference to Python
omit 5c237c0822f Complete the license header of the TS SDK conformance check
omit 36fb91255f3 TS SDK: add triggerDagRun to the native Dag surface
omit 1319665a919 TS SDK: add multi-way branching with dag.switch
omit e911eccfcb4 TS SDK: add conditional branching with dag.if
omit 4ae5fb16161 TS SDK: answer the Dag-parsing request from bundle.serve
omit e392726a3b5 TS SDK: serialize a native Dag to Dag JSON
omit c12ac5a85c0 TS SDK: reject a Dag whose order-only edges form a cycle
omit ada2c65243e TS SDK: add task groups to the native Dag surface
omit 0495ed367c2 TS SDK: draw order-only edges with before and after
omit 542efbf4117 TS SDK: default a task id to the handler's name
add 3fa2b8aad9d TS SDK: default a task id to the handler's name
add a5bb084089a TS SDK: draw order-only edges with before and after
add 5b7b2d5e09f TS SDK: add task groups to the native Dag surface
add 6473db76f0b TS SDK: reject a Dag whose order-only edges form a cycle
add fc27f3164b6 TS SDK: serialize a native Dag to Dag JSON
add 90658e5ff3c TS SDK: answer the Dag-parsing request from bundle.serve
add c1f2661bb98 TS SDK: add conditional branching with dag.if
add e6d7b2de893 TS SDK: add multi-way branching with dag.switch
add 9301fcfb7c5 TS SDK: add triggerDagRun to the native Dag surface
add 32456fb458e Complete the license header of the TS SDK conformance check
add 93f18940610 Java SDK: Describe native Java Dags without reference to Python
add 57ddfadbf87 Java SDK: Generate Dag and task configuration from the serialization schema
add 98d421a45c0 Java SDK: Resolve a task's arguments from the Dag's own wiring
add 65b09deb5d1 Java SDK: Declare a Dag's task graph with @Builder.Deps
add 196ed3af56e Java SDK: Serialize native Dags to DagSerialization v3 on parse requests
new 3801de79ad6 TS SDK: cover native TypeScript Dags end to end
new 75ac609e9bc Parse the native TypeScript Dag in the compose e2e
new bac79752c03 Check the native TypeScript Dag's source and triggered run
new 6707aa83b8c Keep the containerized Java build's Maven repository on the host
new d72a4e218d0 Add a native Java Dag to the compose e2e
new 9e4160d7169 Run the Lang-SDK e2e jobs when native Dag parsing changes
new 9582bdaaf28 Share the Dag run lookup and the native Java queue name
new dedaa3048e3 Build the k8s TypeScript artifacts only for the native Dag test
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (92d2b908f5e)
\
N -- N -- N refs/heads/jason/lang-sdk-e2e/08-native-e2e (dedaa3048e3)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../language-sdks/typescript.rst | 16 +---
ts-sdk/adr/0002-native-dag-interface.md | 35 +++----
ts-sdk/api-docs/dag-authoring-api.ts | 11 +--
ts-sdk/src/cli/pack.ts | 4 +-
ts-sdk/src/coordinator/serde.ts | 2 +-
ts-sdk/src/index.ts | 2 -
ts-sdk/src/sdk/arg-list.ts | 82 -----------------
ts-sdk/src/sdk/dag.ts | 96 +++----------------
ts-sdk/tests/cli/pack.test.ts | 30 ------
ts-sdk/tests/coordinator/arg-binding.test.ts | 23 +----
ts-sdk/tests/coordinator/serde.test.ts | 22 +----
ts-sdk/tests/public-api.test.ts | 31 +------
ts-sdk/tests/sdk/dag.test.ts | 102 +++------------------
13 files changed, 52 insertions(+), 404 deletions(-)
delete mode 100644 ts-sdk/src/sdk/arg-list.ts
(airflow) 01/08: TS SDK: cover native TypeScript Dags end to end
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3801de79ad614b660541975264d0e76e18722e7d
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 21 09:11:04 2026 +0000
TS SDK: cover native TypeScript Dags end to end
(cherry picked from commit 2b76bb3fae1a4c1c5f08ae47d46ec5b0171f4bf6)
---
.../language-sdks/typescript.rst | 6 +
.../airflow_e2e_tests/e2e_test_utils/clients.py | 16 ++
.../ts_sdk_tests/test_ts_sdk_native_dag.py | 209 +++++++++++++++++++++
dev/breeze/doc/images/output_k8s.svg | 8 +-
dev/breeze/doc/images/output_k8s.txt | 2 +-
.../doc/images/output_k8s_setup-lang-sdk-test.svg | 38 ++--
.../doc/images/output_k8s_setup-lang-sdk-test.txt | 2 +-
.../airflow_breeze/commands/kubernetes_commands.py | 165 ++++++++++++++--
.../commands/kubernetes_commands_config.py | 1 +
kubernetes-tests/lang_sdk/Dockerfile.typescript | 41 ++++
kubernetes-tests/lang_sdk/README.md | 17 +-
kubernetes-tests/lang_sdk/config/values.yaml | 12 +-
.../pod_templates/lang_sdk_typescript.yaml | 115 ++++++++++++
.../test_lang_sdk_coordinator_executor.py | 80 ++++++++
ts-sdk/adr/0002-native-dag-interface.md | 18 +-
ts-sdk/example/README.md | 8 +-
ts-sdk/example/src/main.ts | 12 +-
ts-sdk/example/src/native.ts | 163 ++++++++++++++++
ts-sdk/tests/sdk/native-dag-example.test.ts | 207 ++++++++++++++++++++
ts-sdk/tsconfig.json | 10 +-
ts-sdk/vitest.config.ts | 11 ++
21 files changed, 1089 insertions(+), 52 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
index 23f60986224..77726061b5d 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/typescript.rst
@@ -445,6 +445,12 @@ in the task spec.
Templated arguments pass through untouched, so ``{{ ds }}`` in a ``conf`` value is rendered
server-side where rendering already happens.
+A worked example
+~~~~~~~~~~~~~~~~
+
+``ts-sdk/example/src/native.ts`` puts the constructs above into one Dag, registered on the same
+bundle as the mixed-language handlers beside it, so a single artifact serves both authoring modes.
+
``new Dag`` and ``dag.task`` both take a trailing spec of Airflow options:
``{ schedule: "@daily", tags: ["etl"] }`` for the Dag, ``{ retries: 2, retryDelay: 30 }`` for a task.
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
index d56cecae3d7..803edf515e7 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import time
from datetime import datetime, timezone
from functools import cached_property
+from http import HTTPStatus
import boto3
import requests
@@ -163,6 +164,21 @@ class AirflowClient:
"""Get an Airflow Variable via API."""
return self._make_request(method="GET", endpoint=f"variables/{key}")
+ def set_variable(self, key: str, value: str, description: str | None = None):
+ """Create or replace an Airflow Variable via API."""
+ body = {"key": key, "value": value, "description": description}
+ try:
+ return self._make_request(method="POST", endpoint="variables", json=body)
+ except requests.HTTPError as exc:
+ # 409 == it already exists, from an earlier run of the same suite.
+ if exc.response is None or exc.response.status_code != HTTPStatus.CONFLICT:
+ raise
+ return self._make_request(method="PATCH", endpoint=f"variables/{key}", json=body)
+
+ def get_tasks(self, dag_id: str):
+ """List a Dag's tasks, with the edges each one carries."""
+ return self._make_request(method="GET", endpoint=f"dags/{dag_id}/tasks")
+
def trigger_dag_and_wait(self, dag_id: str, json=None):
"""Trigger a DAG and wait for it to complete."""
self.un_pause_dag(dag_id)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
new file mode 100644
index 00000000000..f23684d7a5b
--- /dev/null
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
@@ -0,0 +1,209 @@
+# 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.
+"""
+End-to-end test of a Dag declared entirely in TypeScript.
+
+Run with::
+
+ E2E_TEST_MODE=ts_sdk RUN_TS_SDK_NATIVE_DAG_TESTS=true \\
+ uv run --project airflow-e2e-tests pytest \\
+ tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py -xvs
+
+Unlike ``test_ts_sdk_dag.py``, no Python file declares this Dag: the Dag processor asks the
+``airflow-ts-pack`` bundle to parse itself, and the bundle answers with the serialized Dag that
+``ts-sdk/example/src/native.ts`` built. The graph is a graph rather than a chain -- a task group, a
+named fan-in, order-only edges, a conditional, a multi-way branch, and a ``TriggerDagRunOperator`` a
+Python worker runs.
+
+Gated behind ``RUN_TS_SDK_NATIVE_DAG_TESTS`` because it needs a Dag processor that can dispatch a
+parse request to a language coordinator. Until that lands, the bundle answers a request nothing
+sends, and the Dag never appears.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+import pytest
+
+from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
+
+_RUN_NATIVE = os.environ.get("RUN_TS_SDK_NATIVE_DAG_TESTS", "").lower() in ("true", "1")
+
+# Parsing the bundle launches node before the first task is even scheduled, so
+# allow the same headroom the mixed-language suite does.
+_TS_TASK_TIMEOUT = 600
+
+_DAG_ID = "typescript_native_example"
+
+# Read by the handlers; see ts-sdk/example/src/native.ts.
+_NORTH_ROWS_VARIABLE = "typescript_native_north_rows"
+_SOUTH_ROWS_VARIABLE = "typescript_native_south_rows"
+_CADENCE_VARIABLE = "typescript_native_cadence"
+
+pytestmark = pytest.mark.skipif(
+ not _RUN_NATIVE,
+ reason="Needs a Dag processor that dispatches parse requests to a language coordinator "
+ "(RUN_TS_SDK_NATIVE_DAG_TESTS)",
+)
+
+
+@dataclass
+class _CompletedRun:
+ client: AirflowClient
+ run_id: str
+ state: str
+ ti_states: dict[str, str]
+
+ def xcom(self, task_id: str, key: str = "return_value"):
+ return self.client.get_xcom_value(dag_id=_DAG_ID, task_id=task_id, run_id=self.run_id, key=key).get(
+ "value"
+ )
+
+
+@pytest.fixture(scope="module")
+def parsed_dag() -> AirflowClient:
+ """A client that has waited for the bundle's Dag to be parsed and registered."""
+ client = AirflowClient()
+ # The Dag processor spawns node to parse the bundle, so the Dag appears some
+ # time after the deployment is up; every read below would 404 until then.
+ client.wait_for_dag(_DAG_ID, timeout=_TS_TASK_TIMEOUT)
+ return client
+
+
+@pytest.fixture(scope="module")
+def completed_run(parsed_dag: AirflowClient) -> _CompletedRun:
+ """Trigger the native Dag once, with the inputs every test below reads back."""
+ client = parsed_dag
+ # Both regions non-empty, so the conditional takes its `then` branch, and a
+ # weekly cadence so the branch's choice is known rather than guessed.
+ for key, value in (
+ (_NORTH_ROWS_VARIABLE, "3"),
+ (_SOUTH_ROWS_VARIABLE, "2"),
+ (_CADENCE_VARIABLE, "weekly"),
+ ):
+ client.set_variable(key, value)
+
+ client.un_pause_dag(_DAG_ID)
+ resp = client.trigger_dag(_DAG_ID, json={"logical_date": datetime.now(timezone.utc).isoformat()})
+ run_id = resp["dag_run_id"]
+ state = client.wait_for_dag_run(dag_id=_DAG_ID, run_id=run_id, timeout=_TS_TASK_TIMEOUT)
+ ti_resp = client.get_task_instances(dag_id=_DAG_ID, run_id=run_id)
+ return _CompletedRun(
+ client=client,
+ run_id=run_id,
+ state=state,
+ ti_states={ti["task_id"]: ti.get("state") for ti in ti_resp.get("task_instances", [])},
+ )
+
+
+def test_the_dag_the_bundle_parsed_is_registered(parsed_dag: AirflowClient):
+ """The Dag exists without any Python file declaring it."""
+ dag = parsed_dag.get_dag(_DAG_ID)
+
+ assert dag["dag_id"] == _DAG_ID
+ # The serializer expands a cron preset, so "@daily" is recorded as its expression.
+ assert dag.get("timetable_summary") == "0 0 * * *"
+ # `tags` is a list of objects, each naming one tag.
+ assert {tag["name"] for tag in dag.get("tags") or []} >= {"typescript", "native"}
+
+
+def test_the_graph_carries_every_construct(parsed_dag: AirflowClient):
+ """Group prefixes, the fan-in, the branches and the trigger all survive parsing."""
+ tasks = parsed_dag.get_tasks(_DAG_ID).get("tasks", [])
+ downstream = {task["task_id"]: set(task["downstream_task_ids"]) for task in tasks}
+
+ # The task group prefixed its members.
+ assert {"extract.north", "extract.south"} <= set(downstream)
+ # The named fan-in put both extract tasks upstream of summarize. The API
+ # reports downstream edges only, so upstream is read by inverting them.
+ assert {"extract.north", "extract.south"} <= {
+ task_id for task_id, down in downstream.items() if "summarize" in down
+ }
+ # The conditional and the branch reach their candidates.
+ assert downstream["has_rows"] >= {"load_rows", "report_empty"}
+ assert downstream["pick_cadence"] >= {"publish_daily", "publish_weekly"}
+ # The group edge, expanded onto the tasks the group leaves from.
+ assert {"extract.north", "extract.south"} <= {
+ task_id for task_id, down in downstream.items() if "pick_cadence" in down
+ }
+ # Cleanup sits behind every branch outcome, which is what its trigger rule is for.
+ assert {"load_rows", "report_empty", "publish_daily", "publish_weekly"} <= {
+ task_id for task_id, down in downstream.items() if "cleanup" in down
+ }
+ # And the one task Python runs rather than the Node coordinator.
+ by_id = {task["task_id"]: task for task in tasks}
+ assert by_id["trigger_downstream"]["operator_name"] == "TriggerDagRunOperator"
+
+
+def test_dag_run_succeeded(completed_run: _CompletedRun):
+ assert completed_run.state == "success", (
+ f"expected the run to succeed; got {completed_run.state!r}. task states: {completed_run.ti_states}"
+ )
+
+
+def test_the_taken_branches_ran_and_the_others_skipped(completed_run: _CompletedRun):
+ """A branch is a run-time skip, so the states are what prove it worked."""
+ states = completed_run.ti_states
+
+ # Both regions had rows, so the conditional followed `then`.
+ assert states.get("load_rows") == "success"
+ assert states.get("report_empty") == "skipped"
+
+ # The cadence Variable said weekly, so that is the case the branch chose.
+ assert states.get("publish_weekly") == "success"
+ assert states.get("publish_daily") == "skipped"
+
+
+def test_every_other_task_succeeded(completed_run: _CompletedRun):
+ always_run = [
+ "extract.north",
+ "extract.south",
+ "summarize",
+ "has_rows",
+ "pick_cadence",
+ "cleanup",
+ "trigger_downstream",
+ ]
+ for task_id in always_run:
+ assert completed_run.ti_states.get(task_id) == "success", (
+ f"{task_id!r} did not succeed. all task states: {completed_run.ti_states}"
+ )
+
+
+def test_xcoms_flow_between_typescript_tasks(completed_run: _CompletedRun):
+ """The fan-in's arguments reached the handler, and its own push came back."""
+ assert completed_run.xcom("extract.north") == {"region": "north", "rows": 3}
+ assert completed_run.xcom("extract.south") == {"region": "south", "rows": 2}
+ assert completed_run.xcom("summarize") == {"total": 5, "regions": 2}
+ # Pushed under its own key by the summarize handler, and read back by load_rows.
+ assert completed_run.xcom("summarize", key="region_total") == 5
+ assert completed_run.xcom("load_rows") == {"loaded": 5}
+
+
+def test_the_decisions_are_recorded(completed_run: _CompletedRun):
+ """A condition returns its boolean, and a branch the task id it chose."""
+ assert completed_run.xcom("has_rows") is True
+ assert completed_run.xcom("pick_cadence") == "publish_weekly"
+
+
+def test_a_cleared_skipped_branch_stays_skipped(completed_run: _CompletedRun):
+ """What `_can_skip_downstream` plus the skipmixin XCom are for."""
+ assert completed_run.xcom("has_rows", key="skipmixin_key") == {"skipped": ["report_empty"]}
+ assert completed_run.xcom("pick_cadence", key="skipmixin_key") == {"skipped": ["publish_daily"]}
diff --git a/dev/breeze/doc/images/output_k8s.svg b/dev/breeze/doc/images/output_k8s.svg
index 085b233e8aa..92f835a53e8 100644
--- a/dev/breeze/doc/images/output_k8s.svg
+++ b/dev/breeze/doc/images/output_k8s.svg
@@ -206,10 +206,10 @@
</text><text class="breeze-k8s-r5" x="0" y="727.6" textLength="12.2" clip-path="url(#breeze-k8s-line-29)">│</text><text class="breeze-k8s-r1" x="280.6" y="727.6" textLength="1159" clip-path="url(#breeze-k8s-line-29)">resources, and run the optional per-overlay pytest module.                              &# [...]
</text><text class="breeze-k8s-r5" x="0" y="752" textLength="12.2" clip-path="url(#breeze-k8s-line-30)">│</text><text class="breeze-k8s-r4" x="24.4" y="752" textLength="231.8" clip-path="url(#breeze-k8s-line-30)">run-complete-tests </text><text class="breeze-k8s-r1" x="280.6" y="752" textLength="1159" clip-path="url(#breeze-k8s-line-30)">Run complete k8s tests consisting of: creating cluster, building and uploading image, d [...]
</text><text class="breeze-k8s-r5" x="0" y="776.4" textLength="12.2" clip-path="url(#breeze-k8s-line-31)">│</text><text class="breeze-k8s-r1" x="280.6" y="776.4" textLength="1159" clip-path="url(#breeze-k8s-line-31)">airflow, running tests and deleting clusters (optionally for all clusters in parallel).        </text><text class="breeze-k8s-r5" x="1451.8" y="776.4" textLength="12.2" clip-path=" [...]
-</text><text class="breeze-k8s-r5" x="0" y="800.8" textLength="12.2" clip-path="url(#breeze-k8s-line-32)">│</text><text class="breeze-k8s-r4" x="24.4" y="800.8" textLength="231.8" clip-path="url(#breeze-k8s-line-32)">setup-lang-sdk-test</text><text class="breeze-k8s-r1" x="280.6" y="800.8" textLength="1159" clip-path="url(#breeze-k8s-line-32)">Provision the lang-SDK (Go + Java) coordinator system test on an already-deployed  [...]
-</text><text class="breeze-k8s-r5" x="0" y="825.2" textLength="12.2" clip-path="url(#breeze-k8s-line-33)">│</text><text class="breeze-k8s-r1" x="280.6" y="825.2" textLength="1159" clip-path="url(#breeze-k8s-line-33)">KubernetesExecutor cluster: build artifacts, build + load the Java worker image, deploy        </text><text class="breeze-k8s-r5" x="1451.8" y="825.2" textLength="12.2" clip-path=" [...]
-</text><text class="breeze-k8s-r5" x="0" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-line-34)">│</text><text class="breeze-k8s-r1" x="280.6" y="849.6" textLength="1159" clip-path="url(#breeze-k8s-line-34)">localstack S3, upload artifacts + stub Dag, create config, and upgrade the Helm release. Run   </text><text class="breeze-k8s-r5" x="1451.8" y="849.6" textLength="12.2" clip-path="url(#breez [...]
-</text><text class="breeze-k8s-r5" x="0" y="874" textLength="12.2" clip-path="url(#breeze-k8s-line-35)">│</text><text class="breeze-k8s-r1" x="280.6" y="874" textLength="866.2" clip-path="url(#breeze-k8s-line-35)">the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests </text><text class="breeze-k8s-r4" x="1146.8" y="874" textLength="122" clip-path="url(#breeze-k8s-line-35)">--executor</text><text class="breeze-k8s-r5" x="1451.8" y=" [...]
+</text><text class="breeze-k8s-r5" x="0" y="800.8" textLength="12.2" clip-path="url(#breeze-k8s-line-32)">│</text><text class="breeze-k8s-r4" x="24.4" y="800.8" textLength="231.8" clip-path="url(#breeze-k8s-line-32)">setup-lang-sdk-test</text><text class="breeze-k8s-r1" x="280.6" y="800.8" textLength="1159" clip-path="url(#breeze-k8s-line-32)">Provision the lang-SDK (Go + Java + TypeScript) coordinator system test on an alr [...]
+</text><text class="breeze-k8s-r5" x="0" y="825.2" textLength="12.2" clip-path="url(#breeze-k8s-line-33)">│</text><text class="breeze-k8s-r1" x="280.6" y="825.2" textLength="1159" clip-path="url(#breeze-k8s-line-33)">KubernetesExecutor cluster: build artifacts, build + load the Java and TypeScript worker       </text><text class="breeze-k8s-r5" x="1451.8" y="825.2" textLength="12.2" clip-path="url(# [...]
+</text><text class="breeze-k8s-r5" x="0" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-line-34)">│</text><text class="breeze-k8s-r1" x="280.6" y="849.6" textLength="1159" clip-path="url(#breeze-k8s-line-34)">images, deploy localstack S3, upload artifacts + stub Dag, create config, and upgrade the Helm </text><text class="breeze-k8s-r5" x="1451.8" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-line [...]
+</text><text class="breeze-k8s-r5" x="0" y="874" textLength="12.2" clip-path="url(#breeze-k8s-line-35)">│</text><text class="breeze-k8s-r1" x="280.6" y="874" textLength="1024.8" clip-path="url(#breeze-k8s-line-35)">release. Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests </text><text class="breeze-k8s-r4" x="1305.4" y="874" textLength="122" clip-path="url(#breeze-k8s-line-35)">--executor</text><text class="breez [...]
</text><text class="breeze-k8s-r5" x="0" y="898.4" textLength="12.2" clip-path="url(#breeze-k8s-line-36)">│</text><text class="breeze-k8s-r1" x="280.6" y="898.4" textLength="268.4" clip-path="url(#breeze-k8s-line-36)">KubernetesExecutor -- </text><text class="breeze-k8s-r6" x="549" y="898.4" textLength="24.4" clip-path="url(#breeze-k8s-line-36)">-k</text><text class="breeze-k8s-r1" x="573.4" y="898.4" textLength="866.2" clip-path="url(#breeze-k8s-line-36)"> test_lang_sdk_c [...]
</text><text class="breeze-k8s-r5" x="0" y="922.8" textLength="12.2" clip-path="url(#breeze-k8s-line-37)">│</text><text class="breeze-k8s-r4" x="24.4" y="922.8" textLength="231.8" clip-path="url(#breeze-k8s-line-37)">shell              </text><text class="breeze-k8s-r1" x="280.6" y="922.8" textLength="1159" clip-path="url(#breeze-k8s-line-37)">Run shell environment for the current KinD [...]
</text><text class="breeze-k8s-r5" x="0" y="947.2" textLength="12.2" clip-path="url(#breeze-k8s-line-38)">│</text><text class="breeze-k8s-r4" x="24.4" y="947.2" textLength="231.8" clip-path="url(#breeze-k8s-line-38)">k9s                </text><text class="breeze-k8s-r1" x="280.6" y="947.2" textLength="1159" clip-path="url(#breeze-k8s-line-38)">Run k9s tool. You can pass any  [...]
diff --git a/dev/breeze/doc/images/output_k8s.txt b/dev/breeze/doc/images/output_k8s.txt
index d0c38089093..b983d60ab98 100644
--- a/dev/breeze/doc/images/output_k8s.txt
+++ b/dev/breeze/doc/images/output_k8s.txt
@@ -1 +1 @@
-09b8773fcafec9f720cc6512fa12f73d
+e6bce3a1cc0d3660f16669fe5c4b1338
diff --git a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg
index 55b1d78afae..a9e646da6c4 100644
--- a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg
+++ b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg
@@ -1,4 +1,4 @@
-<svg class="rich-terminal" viewBox="0 0 1482 611.1999999999999" xmlns="http://www.w3.org/2000/svg">
+<svg class="rich-terminal" viewBox="0 0 1482 684.4" xmlns="http://www.w3.org/2000/svg">
<!-- Generated with Rich https://www.textualize.io -->
<style>
@@ -43,7 +43,7 @@
<defs>
<clipPath id="breeze-k8s-setup-lang-sdk-test-clip-terminal">
- <rect x="0" y="0" width="1463.0" height="560.1999999999999" />
+ <rect x="0" y="0" width="1463.0" height="633.4" />
</clipPath>
<clipPath id="breeze-k8s-setup-lang-sdk-test-line-0">
<rect x="0" y="1.5" width="1464" height="24.65"/>
@@ -111,9 +111,18 @@
<clipPath id="breeze-k8s-setup-lang-sdk-test-line-21">
<rect x="0" y="513.9" width="1464" height="24.65"/>
</clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-22">
+ <rect x="0" y="538.3" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-23">
+ <rect x="0" y="562.7" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-24">
+ <rect x="0" y="587.1" width="1464" height="24.65"/>
+ </clipPath>
</defs>
- <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="609.2" rx="8"/><text class="breeze-k8s-setup-lang-sdk-test-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: k8s setup-lang-sdk-test</text>
+ <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="682.4" rx="8"/><text class="breeze-k8s-setup-lang-sdk-test-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: k8s setup-lang-sdk-test</text>
<g transform="translate(26,22)">
<circle cx="0" cy="0" r="7" fill="#ff5f57"/>
<circle cx="22" cy="0" r="7" fill="#febc2e"/>
@@ -126,10 +135,10 @@
<text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="20" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-0)">
</text><text class="breeze-k8s-setup-lang-sdk-test-r2" x="12.2" y="44.4" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-1)">Usage:</text><text class="breeze-k8s-setup-lang-sdk-test-r3" x="97.6" y="44.4" textLength="366" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-1)">breeze k8s setup-lang-sdk-test</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="475.8" y="44.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-1)">[</te [...]
</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="68.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-2)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="93.2" textLength="1415.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">Provision the lang-SDK (Go + Java) coordinator system test on an already-deployed KubernetesExecutor cluster: build </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="93.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="117.6" textLength="1427.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-4)">artifacts, build + load the Java worker image, deploy localstack S3, upload artifacts + stub Dag, create config, and </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="117.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang- [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="142" textLength="1232.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">upgrade the Helm release. Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests </text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="1244.4" y="142" textLength="122" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">--executor</text><text class="b [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="166.4" textLength="268.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">KubernetesExecutor -- </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="280.6" y="166.4" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">-k</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="305" y="166.4" textLength="463.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)"> test_l [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="93.2" textLength="1390.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">Provision the lang-SDK (Go + Java + TypeScript) coordinator system test on an already-deployed KubernetesExecutor </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="93.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="117.6" textLength="1439.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-4)">cluster: build artifacts, build + load the Java and TypeScript worker images, deploy localstack S3, upload artifacts +</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="117.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test- [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="142" textLength="1378.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">stub Dag, create config, and upgrade the Helm release. Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="142" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="166.4" textLength="207.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">breeze k8s tests </text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="219.6" y="166.4" textLength="122" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">--executor</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="166.4" textLength="280.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)"> [...]
</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="190.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-7)">
</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="215.2" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="215.2" textLength="305" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8)"> K8S lang-SDK test flags </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="329.4" y="215.2" textLength="1110.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8 [...]
</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="239.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="239.6" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9)">--python            </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="292.8" y="239.6" textLength="24.4" clip-path="url(#breeze-k8s [...]
@@ -140,12 +149,15 @@
</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="361.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="361.6" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">--java-image        </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="361.6" textLength="1098" clip-path="url(#breeze-k8s-setup-lang-sdk-te [...]
</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="386" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="386" textLength="1098" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">the prod image plus a headless JRE (Dockerfile.java) and loading it into the kind cluster.</text><text class="breeze-k8s-setup-la [...]
</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="410.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="341.6" y="410.4" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">(TEXT)</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="410.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">│</text><text class="breeze-k8s-setup- [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="434.8" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="434.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="459.2" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="459.2" textLength="195.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)"> Common options </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="219.6" y="459.2" textLength="1220" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">───────────── [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="483.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="483.6" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">--verbose</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="483.6" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">-v</text><text class="breeze-k8s-set [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="508" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="508" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">--dry-run</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="508" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">-D</text><text class="breeze-k8s-setup-lan [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="532.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="532.4" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">--help   </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="532.4" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">-h</text><text class= [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="556.8" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="556.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="434.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="434.8" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">--ts-image          </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="434.8" textLength="1098" clip-path="url(#breeze-k8s-setup-l [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="459.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="459.2" textLength="1098" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">building the prod image plus Node (Dockerfile.typescript) and loading it into the kind    </text><text class="breez [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="483.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="483.6" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">cluster. </text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="451.4" y="483.6" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">(TEXT)</text><text class="bree [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="508" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="508" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="532.4" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="532.4" textLength="195.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)"> Common options </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="219.6" y="532.4" textLength="1220" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">───────────── [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="556.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="556.8" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">--verbose</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="556.8" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">-v</text><text class="breeze-k8s-set [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="581.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="581.2" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">--dry-run</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="581.2" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">-D</text><text class="breeze-k8s-set [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="605.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="605.6" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">--help   </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="605.6" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">-h</text><text class= [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="630" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="630" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">
</text>
</g>
</g>
diff --git a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt
index 732dfe90120..c9d173a905d 100644
--- a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt
+++ b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt
@@ -1 +1 @@
-e79130d4518f85b8b1783b9a730f78fa
+6a9a0568a2b313fb5a6780431efd2937
diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
index 564c8fcc1c0..6b2eb800c5b 100644
--- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
+++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
@@ -2505,6 +2505,14 @@ LANG_SDK_GRADLE_CACHE_PATH = AIRFLOW_ROOT_PATH / "files" / "gradle"
# pod_template_file base image.
LANG_SDK_JAVA_WORKER_IMAGE = "lang-sdk-java-worker:latest"
LANG_SDK_JAVA_DOCKERFILE = LANG_SDK_PATH / "Dockerfile.java"
+# The typescript queue needs Node, which the prod image does not carry either.
+# The packed bundle is `ts-sdk/example`, already built by that workspace, so
+# there is no example dir of its own here.
+LANG_SDK_TS_WORKER_IMAGE = "lang-sdk-ts-worker:latest"
+LANG_SDK_TS_DOCKERFILE = LANG_SDK_PATH / "Dockerfile.typescript"
+LANG_SDK_TS_EXAMPLE_PATH = AIRFLOW_ROOT_PATH / "ts-sdk" / "example"
+LANG_SDK_TS_BUILDER_IMAGE = os.environ.get("NODE_BUILDER_IMAGE", "node:22-alpine")
+LANG_SDK_TS_BUNDLE_NAME = "bundle.min.mjs"
LANG_SDK_AWS_CONN_URI = (
"aws://test:test@/?region_name=us-east-1&"
"endpoint_url=http%3A%2F%2Flocalstack.airflow.svc.cluster.local%3A4566"
@@ -2702,6 +2710,93 @@ def _lang_sdk_build_go_bundle(
shutil.copy(output_bin, go_dir / LANG_SDK_GO_BUNDLE_NAME)
+def _lang_sdk_build_ts_bundle(staging: Path, output: Output | None, *, native: bool = False) -> None:
+ """Pack ``ts-sdk/example`` into ``staging/ts-artifacts/bundle.min.mjs``.
+
+ The example depends on the in-repo ts-sdk by a workspace link, so the pack runs against the
+ checkout rather than a scratch copy: unlike Go's module replace, pnpm resolves the link from
+ the workspace root and a copied tree would lose it.
+
+ By default the build runs in an ephemeral Node toolchain container so the host needs no Node
+ install. In ``native`` mode (used in CI, where ``actions/setup-node`` has already provisioned
+ and cached one) it invokes the host toolchain directly.
+ """
+ ts_dir = staging / "ts-artifacts"
+ ts_dir.mkdir(parents=True, exist_ok=True)
+ pack = ["pnpm", "--filter", "apache-airflow-ts-sdk-example", "run", "build"]
+ if native:
+ get_console(output=output).print("[info]Packing the TypeScript bundle with the host Node toolchain")
+ run_command(pack, cwd=AIRFLOW_ROOT_PATH / "ts-sdk", output=output, check=True)
+ else:
+ uid_gid = f"{os.getuid()}:{os.getgid()}"
+ get_console(output=output).print(
+ f"[info]Packing the TypeScript bundle in {LANG_SDK_TS_BUILDER_IMAGE}"
+ )
+ run_command(
+ [
+ "docker",
+ "run",
+ "--rm",
+ "-u",
+ uid_gid,
+ "-v",
+ f"{AIRFLOW_ROOT_PATH}:{AIRFLOW_ROOT_PATH}",
+ "-w",
+ str(AIRFLOW_ROOT_PATH / "ts-sdk"),
+ "-e",
+ "npm_config_cache=/tmp/.npm",
+ "-e",
+ "COREPACK_ENABLE_DOWNLOAD_PROMPT=0",
+ LANG_SDK_TS_BUILDER_IMAGE,
+ "sh",
+ "-c",
+ "corepack enable pnpm && pnpm install --frozen-lockfile && " + " ".join(pack),
+ ],
+ output=output,
+ check=True,
+ )
+ if get_dry_run():
+ return
+ built = LANG_SDK_TS_EXAMPLE_PATH / "dist" / LANG_SDK_TS_BUNDLE_NAME
+ if not built.is_file():
+ raise SystemExit(f"{built} was not produced; check the pack output above")
+ shutil.copyfile(built, ts_dir / LANG_SDK_TS_BUNDLE_NAME)
+
+
+def _lang_sdk_build_ts_worker_image(
+ base_image: str, python: str, kubernetes_version: str, output: Output | None
+) -> str:
+ """Build the prod+Node TypeScript worker image and load it into the kind cluster."""
+ get_console(output=output).print(
+ f"[info]Building TypeScript worker image {LANG_SDK_TS_WORKER_IMAGE} (Node on top of {base_image})"
+ )
+ run_command(
+ [
+ "docker",
+ "build",
+ "--build-arg",
+ f"BASE_IMAGE={base_image}",
+ "-t",
+ LANG_SDK_TS_WORKER_IMAGE,
+ "-f",
+ str(LANG_SDK_TS_DOCKERFILE),
+ str(LANG_SDK_PATH),
+ ],
+ output=output,
+ check=True,
+ )
+ cluster_name = get_kind_cluster_name(python=python, kubernetes_version=kubernetes_version)
+ get_console(output=output).print(f"[info]Loading {LANG_SDK_TS_WORKER_IMAGE} into {cluster_name}")
+ run_command_with_k8s_env(
+ ["kind", "load", "docker-image", "--name", cluster_name, LANG_SDK_TS_WORKER_IMAGE],
+ python=python,
+ kubernetes_version=kubernetes_version,
+ output=output,
+ check=True,
+ )
+ return LANG_SDK_TS_WORKER_IMAGE
+
+
def _lang_sdk_build_java_jar(
staging: Path, java_sdk_source: Path, output: Output | None, *, native: bool = False
) -> None:
@@ -2862,23 +2957,30 @@ def _lang_sdk_upload_artifacts(
).stdout.strip()
go_bundle = staging / "go-artifacts" / "lang_sdk_combined"
+ ts_bundle = staging / "ts-artifacts" / LANG_SDK_TS_BUNDLE_NAME
if get_dry_run():
# The dry-run build steps produce no jar; use the placeholder name so the commands still print.
java_jar = staging / "java-artifacts" / "app.jar"
else:
java_jar = next((staging / "java-artifacts").glob("*.jar"))
stub_dag = LANG_SDK_PATH / "dags" / "lang_sdk_combined.py"
+ # The native TypeScript Dag triggers this one, so it has to be in the
+ # cluster for its trigger task to have a target. It is the same Python Dag
+ # the mixed-language half of `ts-sdk/example` is written against.
+ ts_stub_dag = LANG_SDK_TS_EXAMPLE_PATH / "dags" / "typescript_example.py"
for src, dest in (
(go_bundle, "/tmp/go_bundle"),
(java_jar, "/tmp/app.jar"),
+ (ts_bundle, f"/tmp/{LANG_SDK_TS_BUNDLE_NAME}"),
(stub_dag, "/tmp/lang_sdk_combined.py"),
+ (ts_stub_dag, "/tmp/typescript_example.py"),
):
_lang_sdk_kubectl(
["cp", str(src), f"{HELM_AIRFLOW_NAMESPACE}/{pod}:{dest}"], python, kubernetes_version, output
)
- for bucket in ("go-artifacts", "java-artifacts", "dags"):
+ for bucket in ("go-artifacts", "java-artifacts", "ts-artifacts", "dags"):
_lang_sdk_kubectl(
["exec", "-n", HELM_AIRFLOW_NAMESPACE, pod, "--", "awslocal", "s3", "mb", f"s3://{bucket}"],
python,
@@ -2889,7 +2991,9 @@ def _lang_sdk_upload_artifacts(
uploads = (
("/tmp/go_bundle", "s3://go-artifacts/lang_sdk_combined"),
("/tmp/app.jar", "s3://java-artifacts/app.jar"),
+ (f"/tmp/{LANG_SDK_TS_BUNDLE_NAME}", f"s3://ts-artifacts/{LANG_SDK_TS_BUNDLE_NAME}"),
("/tmp/lang_sdk_combined.py", "s3://dags/lang_sdk_combined.py"),
+ ("/tmp/typescript_example.py", "s3://dags/typescript_example.py"),
)
for src, dest in uploads:
_lang_sdk_kubectl(
@@ -2901,14 +3005,21 @@ def _lang_sdk_upload_artifacts(
def _lang_sdk_apply_configmaps_and_secret(
- python: str, kubernetes_version: str, go_image: str, java_image: str, output: Output | None
+ python: str,
+ kubernetes_version: str,
+ go_image: str,
+ java_image: str,
+ ts_image: str,
+ output: Output | None,
) -> None:
with tempfile.TemporaryDirectory(prefix="lang_sdk_pt_") as tmp:
rendered = Path(tmp)
- for name in ("lang_sdk_golang.yaml", "lang_sdk_java.yaml"):
+ for name in ("lang_sdk_golang.yaml", "lang_sdk_java.yaml", "lang_sdk_typescript.yaml"):
text = (LANG_SDK_PATH / "pod_templates" / name).read_text()
- text = text.replace("__LANG_SDK_GO_IMAGE__", go_image).replace(
- "__LANG_SDK_JAVA_IMAGE__", java_image
+ text = (
+ text.replace("__LANG_SDK_GO_IMAGE__", go_image)
+ .replace("__LANG_SDK_JAVA_IMAGE__", java_image)
+ .replace("__LANG_SDK_TS_IMAGE__", ts_image)
)
(rendered / name).write_text(text)
# Idempotent configmap/secret application via `--dry-run | apply`.
@@ -3067,20 +3178,24 @@ def _setup_lang_sdk_test(
kubernetes_version: str,
go_image: str | None = None,
java_image: str | None = None,
+ ts_image: str | None = None,
output: Output | None = None,
) -> None:
"""Provision the lang-SDK coordinator env on an already-deployed KubernetesExecutor cluster.
- Resolves the go-sdk/java-sdk sources, then builds the Go/Java artifacts, the Java worker
- image and deploys localstack in parallel, then serially uploads the artifacts, applies the
- config + secret, and helm-upgrades Airflow with the lang-SDK values.
+ Resolves the go-sdk/java-sdk sources, then builds the Go/Java/TypeScript artifacts, the Java
+ and TypeScript worker images and deploys localstack in parallel, then serially uploads the
+ artifacts, applies the config + secret, and helm-upgrades Airflow with the lang-SDK values.
"""
go_image = go_image or f"{BuildProdParams(python=python).airflow_image_kubernetes}:latest"
build_java_image = java_image is None
+ build_ts_image = ts_image is None
if java_image is None:
# The worker-image build below produces this fixed tag; resolve it up-front so the config
# rendering (which needs the tag, not the build result) does not depend on the parallel run.
java_image = LANG_SDK_JAVA_WORKER_IMAGE
+ if ts_image is None:
+ ts_image = LANG_SDK_TS_WORKER_IMAGE
# In CI the Go/Java toolchains are provisioned + cached on the host (actions/setup-go, setup-java),
# so building the artifacts natively skips the toolchain-image pulls and reuses the runner caches.
native = os.environ.get("LANG_SDK_NATIVE_TOOLCHAIN", "").lower() == "true"
@@ -3096,6 +3211,10 @@ def _setup_lang_sdk_test(
"Build Java jar",
lambda o: _lang_sdk_build_java_jar(staging, java_sdk_source, o, native=native),
),
+ (
+ "Build TypeScript bundle",
+ lambda o: _lang_sdk_build_ts_bundle(staging, o, native=native),
+ ),
("Deploy localstack", lambda o: _lang_sdk_deploy_localstack(python, kubernetes_version, o)),
]
if build_java_image:
@@ -3105,17 +3224,25 @@ def _setup_lang_sdk_test(
lambda o: _lang_sdk_build_java_worker_image(go_image, python, kubernetes_version, o),
)
)
+ if build_ts_image:
+ steps.append(
+ (
+ "Build TypeScript worker image",
+ lambda o: _lang_sdk_build_ts_worker_image(go_image, python, kubernetes_version, o),
+ )
+ )
_run_lang_sdk_parallel(steps, output=output)
_lang_sdk_upload_artifacts(staging, python, kubernetes_version, output)
- _lang_sdk_apply_configmaps_and_secret(python, kubernetes_version, go_image, java_image, output)
+ _lang_sdk_apply_configmaps_and_secret(python, kubernetes_version, go_image, java_image, ts_image, output)
_lang_sdk_deploy_airflow(python, kubernetes_version, output)
@kubernetes_group.command(
name="setup-lang-sdk-test",
- help="Provision the lang-SDK (Go + Java) coordinator system test on an already-deployed "
- "KubernetesExecutor cluster: build artifacts, build + load the Java worker image, deploy "
- "localstack S3, upload artifacts + stub Dag, create config, and upgrade the Helm release. "
+ help="Provision the lang-SDK (Go + Java + TypeScript) coordinator system test on an "
+ "already-deployed KubernetesExecutor cluster: build artifacts, build + load the Java and "
+ "TypeScript worker images, deploy localstack S3, upload artifacts + stub Dag, create config, "
+ "and upgrade the Helm release. "
"Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests "
"--executor KubernetesExecutor -- -k test_lang_sdk_combined_dag_succeeds`.",
)
@@ -3130,9 +3257,20 @@ def _setup_lang_sdk_test(
help="Image for the Java (JavaCoordinator) worker pod. Must include a JRE. Defaults to building "
"the prod image plus a headless JRE (Dockerfile.java) and loading it into the kind cluster.",
)
+@click.option(
+ "--ts-image",
+ help="Image for the TypeScript (NodeCoordinator) worker pod. Must include Node. Defaults to "
+ "building the prod image plus Node (Dockerfile.typescript) and loading it into the kind cluster.",
+)
@option_verbose
@option_dry_run
-def setup_lang_sdk_test(python: str, kubernetes_version: str, go_image: str | None, java_image: str | None):
+def setup_lang_sdk_test(
+ python: str,
+ kubernetes_version: str,
+ go_image: str | None,
+ java_image: str | None,
+ ts_image: str | None,
+):
result = sync_virtualenv(force_venv_setup=False)
if result.returncode != 0:
sys.exit(result.returncode)
@@ -3142,6 +3280,7 @@ def setup_lang_sdk_test(python: str, kubernetes_version: str, go_image: str | No
kubernetes_version=kubernetes_version,
go_image=go_image,
java_image=java_image,
+ ts_image=ts_image,
output=None,
)
console_print(
diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py
index 919934cc754..b72db432868 100644
--- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py
+++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py
@@ -56,6 +56,7 @@ KUBERNETES_PARAMETERS: dict[str, list[dict[str, str | list[str]]]] = {
"--kubernetes-version",
"--go-image",
"--java-image",
+ "--ts-image",
],
}
],
diff --git a/kubernetes-tests/lang_sdk/Dockerfile.typescript b/kubernetes-tests/lang_sdk/Dockerfile.typescript
new file mode 100644
index 00000000000..1dfda850443
--- /dev/null
+++ b/kubernetes-tests/lang_sdk/Dockerfile.typescript
@@ -0,0 +1,41 @@
+# 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.
+#
+# TypeScript worker image for the "typescript" queue: the stock prod image plus
+# Node.js, so NodeCoordinator can exec `node` on the packed bundle. The Go
+# queue runs on the plain prod image and Java on a JRE image, so each
+# coordinator's queue gets the base image its runtime needs.
+ARG BASE_IMAGE
+FROM ${BASE_IMAGE}
+
+USER root
+# The keyring step pipes curl into gpg, so a failure mid-pipe has to fail the build.
+SHELL ["/bin/bash", "-o", "pipefail", "-c"]
+RUN apt-get update \
+ && apt-get install --no-install-recommends -y ca-certificates curl gnupg \
+ && mkdir -p /etc/apt/keyrings \
+ && curl -fsSL https://deb.nodesource.com/gpgkey/nodesource-repo.gpg.key \
+ | gpg --dearmor -o /etc/apt/keyrings/nodesource.gpg \
+ && echo "deb [signed-by=/etc/apt/keyrings/nodesource.gpg] https://deb.nodesource.com/node_22.x nodistro main" \
+ > /etc/apt/sources.list.d/nodesource.list \
+ && apt-get update \
+ && apt-get install --no-install-recommends -y nodejs \
+ && apt-get purge -y --auto-remove gnupg \
+ && apt-get clean \
+ && rm -rf /var/lib/apt/lists/*
+SHELL ["/bin/sh", "-c"]
+USER airflow
diff --git a/kubernetes-tests/lang_sdk/README.md b/kubernetes-tests/lang_sdk/README.md
index 93583ecbd8b..a9f032a76ed 100644
--- a/kubernetes-tests/lang_sdk/README.md
+++ b/kubernetes-tests/lang_sdk/README.md
@@ -24,6 +24,12 @@ End-to-end test that one Dag mixing **Python + Go + Java** tasks runs to success
`[sdk] coordinators` config. The test lives at
`kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_coordinator_executor.py`.
+The same file also holds `TestNativeTypeScriptDagOnKubernetes`, which runs a Dag with **no Python
+file at all**: the Dag processor asks the packed TypeScript bundle to parse itself. It is gated on
+`RUN_TS_SDK_NATIVE_DAG_K8S_TESTS` as well as `RUN_LANG_SDK_K8S_TESTS`, because dispatching a parse
+request to a language coordinator is not in Airflow yet — the provisioning below is in place, so the
+test becomes real as soon as it is.
+
## How it fits together
```
@@ -57,11 +63,18 @@ coordinator scans.
| `stage_artifacts.py` | Init-container entrypoint; stages an artifact bucket via DagBundle. |
| `pod_templates/lang_sdk_golang.yaml` | `golang` queue worker pod: prod image + go-artifacts init container. |
| `pod_templates/lang_sdk_java.yaml` | `java` queue worker pod: JVM image + java-artifacts init container. |
+| `pod_templates/lang_sdk_typescript.yaml` | `typescript` queue worker pod: Node image + ts-artifacts init container. |
+| `Dockerfile.typescript` | Prod image plus Node.js, which `NodeCoordinator` execs. |
| `manifests/localstack.yaml` | In-cluster S3 (localstack). |
| `config/values.yaml` | Helm overrides: KubernetesExecutor, coordinators (+extra.pod_template_file), queue routing, stub-Dag S3 bundle, AWS conn, scheduler pod-template mount. |
-The Go binary, Java jar, and stub Dag share one object store (localstack) but live in
-**separate buckets** (`go-artifacts`, `java-artifacts`, `dags`).
+The Go binary, Java jar, TypeScript bundle and the Dag files share one object store (localstack)
+but live in **separate buckets** (`go-artifacts`, `java-artifacts`, `ts-artifacts`, `dags`).
+
+The TypeScript bundle is `ts-sdk/example`, packed by `airflow-ts-pack`. It carries both halves of
+that example: the handlers for the Python-declared `typescript_example` Dag, and the natively
+declared `typescript_native_example`. `typescript_example.py` is uploaded to the `dags` bucket too,
+because the native Dag's trigger task targets it.
## Which SDK sources get built
diff --git a/kubernetes-tests/lang_sdk/config/values.yaml b/kubernetes-tests/lang_sdk/config/values.yaml
index a4b5b4004b4..8d782c478f3 100644
--- a/kubernetes-tests/lang_sdk/config/values.yaml
+++ b/kubernetes-tests/lang_sdk/config/values.yaml
@@ -62,13 +62,13 @@ dagProcessor:
aws_conn_id: "aws_localstack"
config:
- # Route the golang/java queues to their coordinators, each carrying an
- # extra.pod_template_file (the worktree-1 feature) so the worker pod is
- # launched from a coordinator-specific template that stages the artifact and,
- # for Java, runs on the JVM-bearing image.
+ # Route the golang/java/typescript queues to their coordinators, each carrying
+ # an extra.pod_template_file so the worker pod is launched from a
+ # coordinator-specific template that stages the artifact and, for Java and
+ # TypeScript, runs on the image carrying that runtime.
sdk:
# Single-line JSON: airflow.cfg is an ini file, so a value with embedded
# newlines would break the rendered configmap.
# yamllint disable-line rule:line-length
- coordinators: '{"go-sdk": {"classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator", "kwargs": {"executables_root": ["/opt/airflow/artifacts/go-artifacts"]}, "extra": {"pod_template_file": "/opt/airflow/lang_sdk/pod_templates/lang_sdk_golang.yaml"}}, "java-sdk": {"classpath": "airflow.sdk.coordinators.java.JavaCoordinator", "kwargs": {"jars_root": ["/opt/airflow/artifacts/java-artifacts"]}, "extra": {"pod_template_file": "/opt/airflow/lang_sdk/pod_templates/lang_sdk_j [...]
- queue_to_coordinator: '{"golang": "go-sdk", "java": "java-sdk"}'
+ coordinators: '{"go-sdk": {"classpath": "airflow.sdk.coordinators.executable.ExecutableCoordinator", "kwargs": {"executables_root": ["/opt/airflow/artifacts/go-artifacts"]}, "extra": {"pod_template_file": "/opt/airflow/lang_sdk/pod_templates/lang_sdk_golang.yaml"}}, "java-sdk": {"classpath": "airflow.sdk.coordinators.java.JavaCoordinator", "kwargs": {"jars_root": ["/opt/airflow/artifacts/java-artifacts"]}, "extra": {"pod_template_file": "/opt/airflow/lang_sdk/pod_templates/lang_sdk_j [...]
+ queue_to_coordinator: '{"golang": "go-sdk", "java": "java-sdk", "typescript": "ts-sdk"}'
diff --git a/kubernetes-tests/lang_sdk/pod_templates/lang_sdk_typescript.yaml b/kubernetes-tests/lang_sdk/pod_templates/lang_sdk_typescript.yaml
new file mode 100644
index 00000000000..1a316f2e7e7
--- /dev/null
+++ b/kubernetes-tests/lang_sdk/pod_templates/lang_sdk_typescript.yaml
@@ -0,0 +1,115 @@
+# 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.
+#
+# Worker pod template for the "typescript" queue (NodeCoordinator).
+#
+# An init container stages the packed ``bundle.min.mjs`` from the localstack
+# ``ts-artifacts`` bucket into a shared emptyDir at /opt/airflow/artifacts via
+# the DagBundle interface (stage_artifacts.py). That path is the coordinator's
+# bundles_root. The base image carries Node.js (Dockerfile.typescript), which the plain
+# prod image does not. Image placeholders are rendered by the
+# ``breeze k8s setup-lang-sdk-test`` command.
+---
+kind: Pod
+apiVersion: v1
+metadata:
+ name: placeholder-name-dont-delete
+ namespace: placeholder-name-dont-delete
+spec:
+ initContainers:
+ - name: stage-ts-bundle
+ image: __LANG_SDK_TS_IMAGE__
+ imagePullPolicy: IfNotPresent
+ command: ["python", "/opt/airflow/lang_sdk/stage_artifacts.py"]
+ env:
+ - name: AIRFLOW_HOME
+ value: /opt/airflow
+ - name: ARTIFACT_BUNDLE_NAME
+ value: ts-artifacts
+ - name: AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_STORAGE_PATH
+ value: /opt/airflow/artifacts
+ - name: AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST
+ value: >-
+ [{"name": "ts-artifacts",
+ "classpath": "airflow.providers.amazon.aws.bundles.s3.S3DagBundle",
+ "kwargs": {"bucket_name": "ts-artifacts", "aws_conn_id": "aws_localstack"}}]
+ - name: AIRFLOW_CONN_AWS_LOCALSTACK
+ valueFrom:
+ secretKeyRef:
+ name: lang-sdk-aws-conn
+ key: uri
+ volumeMounts:
+ - name: lang-sdk-artifacts
+ mountPath: /opt/airflow/artifacts
+ - name: lang-sdk-scripts
+ readOnly: true
+ mountPath: /opt/airflow/lang_sdk
+ containers:
+ - name: base
+ image: __LANG_SDK_TS_IMAGE__
+ env:
+ - name: AIRFLOW__CORE__EXECUTOR
+ value: LocalExecutor
+ - name: AIRFLOW_HOME
+ value: /opt/airflow
+ - name: AIRFLOW__CORE__DAGS_FOLDER
+ value: /opt/airflow/dags
+ - name: AIRFLOW__CORE__FERNET_KEY
+ valueFrom:
+ secretKeyRef:
+ name: airflow-fernet-key
+ key: fernet-key
+ - name: AIRFLOW__DATABASE__SQL_ALCHEMY_CONN
+ valueFrom:
+ secretKeyRef:
+ name: airflow-metadata
+ key: connection
+ volumeMounts:
+ - name: airflow-logs
+ mountPath: /opt/airflow/logs
+ - name: lang-sdk-artifacts
+ mountPath: /opt/airflow/artifacts
+ - name: airflow-config
+ readOnly: true
+ mountPath: /opt/airflow/airflow.cfg
+ subPath: airflow.cfg
+ - name: airflow-config
+ readOnly: true
+ mountPath: /opt/airflow/config/airflow_local_settings.py
+ subPath: airflow_local_settings.py
+ imagePullPolicy: IfNotPresent
+ volumes:
+ - name: airflow-logs
+ emptyDir: {}
+ - name: lang-sdk-artifacts
+ emptyDir: {}
+ - name: lang-sdk-scripts
+ configMap:
+ name: lang-sdk-scripts
+ defaultMode: 420
+ - name: airflow-config
+ configMap:
+ name: airflow-config
+ defaultMode: 420
+ restartPolicy: Never
+ terminationGracePeriodSeconds: 30
+ serviceAccountName: airflow-worker-kubernetes
+ serviceAccount: airflow-worker-kubernetes
+ securityContext:
+ runAsUser: 50000
+ fsGroup: 50000
+ schedulerName: default-scheduler
diff --git a/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_coordinator_executor.py b/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_coordinator_executor.py
index 813329d4b8a..7bde81fbf32 100644
--- a/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_coordinator_executor.py
+++ b/kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_coordinator_executor.py
@@ -38,6 +38,20 @@ from kubernetes_tests.test_base import EXECUTOR, BaseK8STest
_RUN_LANG_SDK = os.environ.get("RUN_LANG_SDK_K8S_TESTS", "").lower() in ("true", "1")
DAG_ID = "lang_sdk_combined"
+# Declared entirely in TypeScript: the Dag processor asks the bundle to parse
+# itself, so no Python file in the bundles folder mentions it.
+NATIVE_TS_DAG_ID = "typescript_native_example"
+NATIVE_TS_TASK_IDS = [
+ "extract.north",
+ "extract.south",
+ "summarize",
+ "has_rows",
+ "load_rows",
+ "pick_cadence",
+ "publish_weekly",
+ "cleanup",
+ "trigger_downstream",
+]
TASK_IDS = [
"python_task_1",
"go_extract",
@@ -50,6 +64,10 @@ TASK_IDS = [
# artifact + start a coordinator subprocess, so allow generous headroom.
_TIMEOUT = 600
+# The native Dag additionally needs a Dag processor that dispatches a parse
+# request to the Node coordinator; until that lands the Dag never appears.
+_RUN_NATIVE_TS = os.environ.get("RUN_TS_SDK_NATIVE_DAG_K8S_TESTS", "").lower() in ("true", "1")
+
@pytest.mark.skipif(
EXECUTOR != "KubernetesExecutor" or not _RUN_LANG_SDK,
@@ -86,3 +104,65 @@ class TestLangSdkCoordinatorExecutor(BaseK8STest):
expected_final_state="success",
timeout=_TIMEOUT,
)
+
+
+@pytest.mark.skipif(
+ EXECUTOR != "KubernetesExecutor" or not _RUN_NATIVE_TS,
+ reason="Runs only on KubernetesExecutor with a Dag processor that parses Lang SDK bundles "
+ "(RUN_TS_SDK_NATIVE_DAG_K8S_TESTS)",
+)
+class TestNativeTypeScriptDagOnKubernetes(BaseK8STest):
+ """The same Dag the SDK and Airflow e2e suites run, under the Node coordinator.
+
+ Covers the path a Dag with no Python file takes: the Dag processor parses the
+ ``airflow-ts-pack`` bundle through the coordinator, the scheduler reads the serialized Dag,
+ and each task runs in its own pod. The graph is a real one -- a task group, a named fan-in,
+ order-only edges, a conditional and a multi-way branch -- so pod-per-task scheduling is
+ exercised against branches that skip rather than a linear chain.
+ """
+
+ def _ensure_variable(self, key: str, value: str) -> None:
+ resp = self.session.post(f"http://{self.host}/variables", json={"key": key, "value": value})
+ assert resp.status_code in (200, 201, 409), f"Could not create variable {key}: {resp.text}"
+
+ @pytest.mark.execution_timeout(1200)
+ def test_native_typescript_dag_succeeds(self):
+ # Both regions non-empty, so the conditional takes its `then` branch and
+ # `report_empty` is the side that skips. The cadence names the case the
+ # multi-way branch chooses, so `publish_daily` is the side that skips.
+ self._ensure_variable("typescript_native_north_rows", "3")
+ self._ensure_variable("typescript_native_south_rows", "2")
+ self._ensure_variable("typescript_native_cadence", "weekly")
+
+ dag_run_id, logical_date = self.start_job_in_kubernetes(NATIVE_TS_DAG_ID, self.host)
+ print(f"Triggered {NATIVE_TS_DAG_ID} run {dag_run_id} (logical_date={logical_date})")
+
+ for task_id in NATIVE_TS_TASK_IDS:
+ self.monitor_task(
+ host=self.host,
+ dag_run_id=dag_run_id,
+ dag_id=NATIVE_TS_DAG_ID,
+ task_id=task_id,
+ expected_final_state="success",
+ timeout=_TIMEOUT,
+ )
+
+ # The branch that was not taken is skipped, not failed, which is what
+ # proves the skip reached the supervisor rather than the task erroring.
+ for skipped in ("report_empty", "publish_daily"):
+ self.monitor_task(
+ host=self.host,
+ dag_run_id=dag_run_id,
+ dag_id=NATIVE_TS_DAG_ID,
+ task_id=skipped,
+ expected_final_state="skipped",
+ timeout=_TIMEOUT,
+ )
+
+ self.ensure_dag_expected_state(
+ host=self.host,
+ logical_date=logical_date,
+ dag_id=NATIVE_TS_DAG_ID,
+ expected_final_state="success",
+ timeout=_TIMEOUT,
+ )
diff --git a/ts-sdk/adr/0002-native-dag-interface.md b/ts-sdk/adr/0002-native-dag-interface.md
index 7c08f4f6009..94c798bcb4d 100644
--- a/ts-sdk/adr/0002-native-dag-interface.md
+++ b/ts-sdk/adr/0002-native-dag-interface.md
@@ -21,7 +21,7 @@
## Status
-Proposed. Revised after the review on #72047.
+Accepted. Revised after the review on #72047, and again as it was implemented.
## Decision
@@ -206,13 +206,17 @@ Dag is read.
- One authoring surface (`dag.task()` plus its factory) covers the graph and each task's arguments,
and `before`/`after` cover edges that carry nothing.
- Handlers are unit-testable as plain functions of their data, with no SDK fixture to construct.
-- `DagSpec` and `TaskSpec` are empty placeholders today — `Record<string, never>`
- (`ts-sdk/src/sdk/dag.ts`), so `new Dag("d", { schedule: "@daily" })` is currently a compile error
- by design. Native declaration is what fills them, generated from the serialized-Dag JSON schema the
- way `src/generated/supervisor.ts` is. This ADR does not choose those fields; it fixes where an
- author writes them.
+- `DagSpec` and `TaskSpec` carry the fields of Airflow's own serialized-Dag JSON schema, generated
+ from it the way `src/generated/supervisor.ts` is, so every field is optional and a misspelled one is
+ a compile error. This ADR does not choose those fields; it fixes where an author writes them. `queue`
+ is the one hand-written Dag field: the schema has no Dag-level queue, but every task of a native Dag
+ runs on the same coordinator, so the queue that routes them there belongs on the Dag.
- `TaskOptions` carries the task's spec and nothing else: the names on the wire are the keys of the
- call itself. With wiring moved to the factory call, `inputs` is no longer an option.
+ call itself. With wiring moved to the factory call, `inputs` is no longer an option, so the trailing
+ argument is the spec's own fields.
+- A native Dag's tasks serialize as stub tasks (`is_stub`, with `_arg_bindings`), which is how the API
+ server resolves each argument per task instance and hands it to a foreign runtime. So the wiring an
+ author writes is the same mechanism a `@task.stub` call already uses, rather than a second one.
- `TaskHandlerArgs` is removed from the public API, `DagRegistry` becomes `Bundle`, and
`serveDags(registry)` becomes `bundle.serve()`, which breaks
0.1.0-beta1 authors; see [ADR-0001](0001-mixed-lang-dag-interface.md) for the shipped call sites
diff --git a/ts-sdk/example/README.md b/ts-sdk/example/README.md
index 94c2ead3951..a51e089cdb5 100644
--- a/ts-sdk/example/README.md
+++ b/ts-sdk/example/README.md
@@ -25,7 +25,13 @@ This example shows the coordinator-mode shape for TypeScript task handlers:
- `src/main.ts` and `src/taskflow.ts` register a `TaskHandler` per stub task and start the coordinator runtime.
One bundle provides for both Dags, and both declare a task called `build_message`.
A handler binds the `(dag_id, task_id)` pair, so the two are different tasks with different bodies.
-- `dist/bundle.min.mjs` is the generated Node.js bundle that Airflow launches.
+- `src/native.ts` declares a third Dag, `typescript_native_example`, **entirely in TypeScript** — no
+ Python file declares its graph. Its schedule, task options and edges are all written on this side,
+ and the bundle answers the Dag processor's parse request with the serialized Dag. The graph is a
+ graph rather than a chain: a task group, a named fan-in, order-only edges, a conditional, a
+ multi-way branch, and a `triggerDagRun` task a Python worker executes.
+- `dist/bundle.min.mjs` is the generated Node.js bundle that Airflow launches. One artifact serves both
+ authoring modes.
The build uses the SDK's `airflow-ts-pack` tool, which bundles the entrypoint
with esbuild and embeds the Airflow metadata generated from the bundle's
diff --git a/ts-sdk/example/src/main.ts b/ts-sdk/example/src/main.ts
index fa97d3f9899..14e68ab2b9e 100644
--- a/ts-sdk/example/src/main.ts
+++ b/ts-sdk/example/src/main.ts
@@ -17,13 +17,15 @@
* under the License.
*/
-// The bundle entry point: one bundle, two Python-owned Dags.
+// The bundle entry point: one bundle, three Dags, both authoring modes.
//
-// Both Dags in `dags/` are declared in Python with `@task.stub` tasks routed to the Node
-// coordinator, so this side only supplies the task bodies.
+// The two Dags in `dags/` are declared in Python with `@task.stub` tasks routed to the Node
+// coordinator, so this side only supplies their task bodies. The third, in `native.ts`, is
+// declared in TypeScript outright — graph, schedule and all.
import { Bundle, getClient, getContext, TaskHandler } from "apache-airflow-ts-sdk";
+import { dag as nativeDag } from "./native.js";
import { buildSummaryMessage, report, summarize } from "./taskflow.js";
export async function buildMessage() {
@@ -72,8 +74,12 @@ export async function readConnection() {
// One register call lists everything this bundle provides.
// `build_message` appears under both Dags: two different handlers, told apart by the dag_id each
// is bound to and never by the task_id alone.
+//
+// One bundle serves both authoring modes: handlers for the Dags a Python file declares, and the
+// natively declared Dag from `native.ts`, whose graph this side owns outright.
const bundle = new Bundle();
bundle.register(
+ nativeDag,
new TaskHandler("typescript_example", "build_message", buildMessage),
new TaskHandler("typescript_example", "read_connection", readConnection),
new TaskHandler("typescript_example", "write_and_delete_variable", writeAndDeleteVariable),
diff --git a/ts-sdk/example/src/native.ts b/ts-sdk/example/src/native.ts
new file mode 100644
index 00000000000..64b8efb2192
--- /dev/null
+++ b/ts-sdk/example/src/native.ts
@@ -0,0 +1,163 @@
+/*!
+ * 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.
+ */
+
+// A Dag declared entirely in TypeScript: no Python file declares its graph.
+//
+// The counterpart of `main.ts`, which supplies handlers for a Dag that a Python
+// file declares. Here the schedule, the tasks, their options and the edges
+// between them are all written on this side, and the bundle answers the Dag
+// processor's parse request with the serialized Dag.
+//
+// The graph is a graph rather than a chain, so the e2e suites exercise the
+// constructs against each other: a task group, a named fan-in, order-only
+// edges, a conditional, a multi-way branch, and a task that triggers another
+// Dag's run.
+//
+// `main.ts` registers this Dag on the same bundle as the mixed-language
+// handlers, so one artifact covers both authoring modes.
+
+import { Dag, getClient } from "apache-airflow-ts-sdk";
+
+/** The regions this example works over. */
+const REGIONS = ["north", "south"] as const;
+
+interface RegionRows {
+ readonly region: string;
+ readonly rows: number;
+}
+
+export const dag = new Dag("typescript_native_example", {
+ schedule: "@daily",
+ catchup: false,
+ tags: ["typescript", "native"],
+ description: "A Dag whose graph, schedule and task options are all declared in TypeScript.",
+ // Routes every task of this Dag to the Node coordinator; see the
+ // `queue_to_coordinator` entry in the TypeScript SDK docs.
+ queue: "typescript",
+});
+
+// --- Extraction, inside a task group -------------------------------------
+//
+// A group prefixes the ids of everything in it, so these become
+// `extract.north` and `extract.south`.
+
+const extract = dag.taskGroup("extract");
+
+const extractNorth = extract.task("north", async (): Promise<RegionRows> => {
+ const configured = await getClient().getVariable("typescript_native_north_rows");
+ return { region: "north", rows: Number(configured ?? 3) };
+});
+
+const extractSouth = extract.task("south", async (): Promise<RegionRows> => {
+ const configured = await getClient().getVariable("typescript_native_south_rows");
+ return { region: "south", rows: Number(configured ?? 0) };
+});
+
+// --- A named fan-in -------------------------------------------------------
+//
+// Two upstreams feed one task under the argument names its handler declares,
+// which is what the wiring object is for.
+
+const summarize = dag.task(
+ "summarize",
+ async ({ north, south }: { north: RegionRows; south: RegionRows }) => {
+ const total = north.rows + south.rows;
+ await getClient().setXCom({ key: "region_total", value: total });
+ return { total, regions: REGIONS.length };
+ },
+);
+
+const summarized = summarize({ north: extractNorth(), south: extractSouth() });
+
+// --- A conditional --------------------------------------------------------
+//
+// The guarded tasks take no parameter for the control edge: the condition's
+// boolean is a run-time signal rather than data, so it is never an argument.
+
+const loadRows = dag.task("load_rows", async () => {
+ const total = await getClient().getXCom<number>({ key: "region_total", taskId: "summarize" });
+ return { loaded: total ?? 0 };
+});
+
+const reportEmpty = dag.task("report_empty", async () => ({ loaded: 0 }));
+
+const loaded = loadRows();
+const reportedEmpty = reportEmpty();
+
+// The upstream is an input like any other, named after the argument it
+// supplies.
+const hasRows = dag.task(
+ "has_rows",
+ async ({ summary }: { summary: { total: number } }) => summary.total > 0,
+);
+const gated = hasRows({ summary: summarized });
+
+dag.if(gated).then(loaded).else(reportedEmpty);
+
+// --- A multi-way branch ---------------------------------------------------
+//
+// A case is the reference the SDK handed back, so renaming a handler cannot
+// silently rewire the Dag, and the compiler checks the candidate exists.
+
+const publishDaily = dag.task("publish_daily", async () => ({ cadence: "daily" }));
+const publishWeekly = dag.task("publish_weekly", async () => ({ cadence: "weekly" }));
+
+const daily = publishDaily();
+const weekly = publishWeekly();
+
+// Decided from a Variable rather than from an upstream value, so the decider
+// takes no argument and the extract group is what orders it.
+const pickCadence = dag.task("pick_cadence", async () => {
+ const cadence = await getClient().getVariable("typescript_native_cadence");
+ return cadence === "weekly" ? weekly : daily;
+});
+const picked = pickCadence();
+
+dag.switch(picked).case(daily).case(weekly);
+
+// --- Order-only edges -----------------------------------------------------
+//
+// `cleanup` follows whichever side of each branch ran, but takes nothing from
+// it, so the edges are drawn between the references rather than through an
+// argument. Its `all_done` rule is what lets it run when a branch skipped the
+// other side.
+//
+// `extract.before(picked)` is the same kind of edge with a group at one end:
+// the decider reads no upstream value, so this is what puts it after every
+// task the group holds.
+
+const cleanup = dag.task("cleanup", async () => undefined, { triggerRule: "all_done" });
+const cleaned = cleanup();
+
+cleaned.after(loaded, reportedEmpty, daily, weekly);
+extract.before(picked);
+
+// --- A task Python runs ---------------------------------------------------
+//
+// A trigger wraps no TypeScript function: it serializes as
+// `TriggerDagRunOperator` and a Python worker executes it, so it inherits no
+// queue from this Dag and needs nothing to call.
+
+dag
+ .triggerDagRun({
+ taskId: "trigger_downstream",
+ dagId: "typescript_example",
+ conf: { triggered_by: "{{ dag.dag_id }}" },
+ })
+ .after(cleaned);
diff --git a/ts-sdk/tests/sdk/native-dag-example.test.ts b/ts-sdk/tests/sdk/native-dag-example.test.ts
new file mode 100644
index 00000000000..72b0353eb83
--- /dev/null
+++ b/ts-sdk/tests/sdk/native-dag-example.test.ts
@@ -0,0 +1,207 @@
+/*!
+ * 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.
+ */
+
+// The shipped native Dag example, built and serialized.
+//
+// The unit suites cover each construct on a Dag written for that construct
+// alone; this reads the example an author is pointed at, so the graph the docs
+// describe is the graph the SDK produces. The Airflow and Kubernetes suites
+// run the same Dag end to end.
+
+import { describe, expect, it } from "vitest";
+import { serializeDag } from "../../src/coordinator/serde.js";
+import { Bundle, listBundleNativeDags } from "../../src/sdk/bundle.js";
+import { finalizeDag, getDagOrderEdges, type Dag } from "../../src/sdk/dag.js";
+import { dag } from "../../example/src/native.js";
+
+type Json = Record<string, unknown>;
+
+const FILELOC = "/bundles/example/bundle.min.mjs";
+
+/** The serialized Dag, and its tasks keyed by task id. */
+function serialized(target: Dag) {
+ const json = serializeDag(target, FILELOC, "bundle.min.mjs") as Json;
+ const tasks = json["tasks"] as { __var: Json }[];
+ return { json, tasks: new Map(tasks.map(({ __var }) => [__var["task_id"] as string, __var])) };
+}
+
+describe("the native Dag example", () => {
+ it("declares its schedule, options and queue in TypeScript", () => {
+ expect(dag.dagId).toBe("typescript_native_example");
+ expect(dag.spec).toMatchObject({
+ schedule: "@daily",
+ catchup: false,
+ tags: ["typescript", "native"],
+ queue: "typescript",
+ });
+ });
+
+ it("builds a graph, not a chain", () => {
+ expect([...dag.taskIds].sort()).toEqual([
+ "cleanup",
+ "extract.north",
+ "extract.south",
+ "has_rows",
+ "load_rows",
+ "pick_cadence",
+ "publish_daily",
+ "publish_weekly",
+ "report_empty",
+ "summarize",
+ "trigger_downstream",
+ ]);
+ });
+
+ it("is fully laid out, so reading it raises nothing", () => {
+ expect(() => finalizeDag(dag)).not.toThrow();
+ });
+
+ it("is served by the same bundle as the mixed-language handlers", () => {
+ // What `main.ts` builds: importing it would start the runtime, so the
+ // registration is rebuilt here rather than imported.
+ const bundle = new Bundle(dag);
+
+ expect(listBundleNativeDags(bundle).map((served) => served.dagId)).toEqual([
+ "typescript_native_example",
+ ]);
+ });
+
+ describe("serializes", () => {
+ it("a named fan-in as an edge from each upstream into one task", () => {
+ const { tasks } = serialized(dag);
+
+ expect(tasks.get("extract.north")?.["downstream_task_ids"]).toContain("summarize");
+ expect(tasks.get("extract.south")?.["downstream_task_ids"]).toContain("summarize");
+ expect(tasks.get("summarize")?.["downstream_task_ids"]).toEqual(["has_rows"]);
+ });
+
+ it("the fan-in's arguments as the bindings the API server resolves", () => {
+ expect(serialized(dag).tasks.get("summarize")?.["_arg_bindings"]).toEqual([
+ { name: "north", kind: "xcom", task_id: "extract.north" },
+ { name: "south", kind: "xcom", task_id: "extract.south" },
+ ]);
+ });
+
+ it("the positional call's argument as one binding on the upstream", () => {
+ // The label is whatever named the parameter, which only the packer can
+ // read; the order and the upstream are what bind.
+ expect(serialized(dag).tasks.get("has_rows")?.["_arg_bindings"]).toMatchObject([
+ { kind: "xcom", task_id: "summarize" },
+ ]);
+ });
+
+ it("the task group, with its tasks prefixed and nested under it", () => {
+ const root = serialized(dag).json["task_group"] as Json;
+ const [kind, group] = (root["children"] as Json)["extract"] as [string, Json];
+
+ expect(kind).toBe("taskgroup");
+ expect(group).toMatchObject({
+ _group_id: "extract",
+ children: {
+ "extract.north": ["operator", "extract.north"],
+ "extract.south": ["operator", "extract.south"],
+ },
+ });
+ });
+
+ it("the group edge, expanded onto the tasks the group leaves from", () => {
+ // `extract.before(picked)` is drawn against the group, and reaches the
+ // task graph through the group's leaves — both extract tasks, since
+ // neither runs after the other inside it.
+ expect(getDagOrderEdges(dag)).toContainEqual({
+ upstream: "extract",
+ downstream: "pick_cadence",
+ });
+
+ const { tasks } = serialized(dag);
+ expect(tasks.get("extract.north")?.["downstream_task_ids"]).toContain("pick_cadence");
+ expect(tasks.get("extract.south")?.["downstream_task_ids"]).toContain("pick_cadence");
+ });
+
+ it("the conditional's control edges as ordinary edges, and its skip marker", () => {
+ // Nothing names the branch in the Dag JSON: the decision is a run-time
+ // skip, and `_can_skip_downstream` is what makes a cleared branch stay
+ // skipped.
+ const { tasks } = serialized(dag);
+ const hasRows = tasks.get("has_rows")!;
+
+ expect(hasRows["downstream_task_ids"]).toEqual(["load_rows", "report_empty"]);
+ expect(hasRows["_can_skip_downstream"]).toBe(true);
+ expect(tasks.get("load_rows")?.["downstream_task_ids"]).toEqual(["cleanup"]);
+ });
+
+ it("the multi-way branch's cases as ordinary edges, and its skip marker", () => {
+ const pick = serialized(dag).tasks.get("pick_cadence")!;
+
+ expect(pick["downstream_task_ids"]).toEqual(["publish_daily", "publish_weekly"]);
+ expect(pick["_can_skip_downstream"]).toBe(true);
+ });
+
+ it("cleanup behind every branch outcome, which is what its all_done rule is for", () => {
+ const { tasks } = serialized(dag);
+
+ for (const outcome of ["load_rows", "report_empty", "publish_daily", "publish_weekly"]) {
+ expect(tasks.get(outcome)?.["downstream_task_ids"]).toContain("cleanup");
+ }
+ expect(tasks.get("cleanup")?.["trigger_rule"]).toBe("all_done");
+ expect(tasks.get("cleanup")?.["downstream_task_ids"]).toEqual(["trigger_downstream"]);
+ });
+
+ it("the trigger task as the Python operator a Python worker runs", () => {
+ const trigger = serialized(dag).tasks.get("trigger_downstream")!;
+
+ expect(trigger).toMatchObject({
+ task_type: "TriggerDagRunOperator",
+ _task_module: "airflow.providers.standard.operators.trigger_dagrun",
+ trigger_dag_id: "typescript_example",
+ });
+ // No TypeScript marker, and no queue of this Dag's, so a Python worker
+ // can pick it up.
+ expect(trigger).not.toHaveProperty("language");
+ expect(trigger).not.toHaveProperty("queue");
+ });
+
+ it("the triggered Dag as a dependency of this one", () => {
+ expect(serialized(dag).json["dag_dependencies"]).toEqual([
+ {
+ source: "typescript_native_example",
+ target: "typescript_example",
+ label: "trigger_downstream",
+ dependency_type: "trigger",
+ dependency_id: "trigger_downstream",
+ },
+ ]);
+ });
+
+ it("every other task as one this runtime executes, on the Dag's queue", () => {
+ const { tasks } = serialized(dag);
+ const typescript = [...tasks.values()].filter((task) => task["language"] === "typescript");
+
+ expect(typescript).toHaveLength(dag.taskIds.length - 1);
+ expect(typescript.every((task) => task["queue"] === "typescript")).toBe(true);
+ });
+
+ it("the schedule as the expanded cron the scheduler rebuilds", () => {
+ expect(serialized(dag).json["timetable"]).toMatchObject({
+ __type: "airflow.timetables.trigger.CronTriggerTimetable",
+ __var: { expression: "0 0 * * *" },
+ });
+ });
+ });
+});
diff --git a/ts-sdk/tsconfig.json b/ts-sdk/tsconfig.json
index a47f167e378..14bba01027d 100644
--- a/ts-sdk/tsconfig.json
+++ b/ts-sdk/tsconfig.json
@@ -23,10 +23,18 @@
// narrows `include` back to src, so none of this reaches the package.
"allowJs": true,
+ // `example/` imports the package by name, which would otherwise resolve to
+ // the built `dist/`. A test that reached into both would hold two copies of
+ // the SDK, whose classes are not each other's. vitest.config.ts carries the
+ // matching alias so the compiler and the runner agree.
+ "paths": {
+ "apache-airflow-ts-sdk": ["./src/index.ts"]
+ },
+
"noEmit": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
- "include": ["src/**/*.ts", "tests/**/*.ts", "scripts/**/*.mjs"]
+ "include": ["src/**/*.ts", "tests/**/*.ts", "scripts/**/*.mjs", "example/src/**/*.ts"]
}
diff --git a/ts-sdk/vitest.config.ts b/ts-sdk/vitest.config.ts
index 723796e6d29..c1ea185f655 100644
--- a/ts-sdk/vitest.config.ts
+++ b/ts-sdk/vitest.config.ts
@@ -17,9 +17,20 @@
* under the License.
*/
+import { fileURLToPath } from "node:url";
+
import { defineConfig } from "vitest/config";
export default defineConfig({
+ resolve: {
+ alias: {
+ // `example/` imports the package by name, which resolves to the built
+ // `dist/`. A test that reaches into both would hold two copies of the
+ // SDK, and a Dag from one is not an instance of the other's class.
+ // Tests run against the sources, so the example does too.
+ "apache-airflow-ts-sdk": fileURLToPath(new URL("./src/index.ts", import.meta.url)),
+ },
+ },
test: {
include: ["tests/**/*.test.ts", "scripts/**/*.test.mjs"],
},
(airflow) 05/08: Add a native Java Dag to the compose e2e
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d72a4e218d053a8ac24227a3ae83c4b10770f6ee
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 08:01:45 2026 +0000
Add a native Java Dag to the compose e2e
A dedicated bundle declares two Dags in Java, one with the interface API
and one with annotations, and routes every task to a "java-native" queue.
Its JAR goes into the Dag bundle, where a root-less Java coordinator parses
it in the Dag processor and runs its tasks. The test checks the edges, the
queues, the run, the XComs and the Dag source the JAR embeds.
---
airflow-e2e-tests/docker/java.yml | 10 +-
airflow-e2e-tests/java-native-bundle/.gitignore | 2 +
airflow-e2e-tests/java-native-bundle/build.gradle | 50 +++++++
.../java-native-bundle/gradle.properties | 20 +++
.../java-native-bundle/settings.gradle | 33 +++++
.../apache/airflow/e2e/NativeBundleBuilder.java | 84 +++++++++++
.../airflow/e2e/nativedag/AnnotationDag.java | 62 +++++++++
.../tests/airflow_e2e_tests/conftest.py | 23 +++-
.../tests/airflow_e2e_tests/constants.py | 5 +
.../java_sdk_tests/test_java_sdk_native_dag.py | 153 +++++++++++++++++++++
10 files changed, 434 insertions(+), 8 deletions(-)
diff --git a/airflow-e2e-tests/docker/java.yml b/airflow-e2e-tests/docker/java.yml
index db4897f44c8..e91ef21f030 100644
--- a/airflow-e2e-tests/docker/java.yml
+++ b/airflow-e2e-tests/docker/java.yml
@@ -22,14 +22,18 @@
# the pre-built bundle JARs (the Java example under /opt/airflow/java-jars, the
# Scala Spark example under /opt/airflow/scala-jars, and the runner-behaviour
# test fixtures under /opt/airflow/java-test-jars), and configures the worker to
-# consume the "java", "scala", and "java-test" Celery queues where @task.stub
-# tasks are routed.
+# consume the "java", "scala", "java-test", and "java-native" Celery queues
+# where Java tasks are routed. The native-Dag JAR is in the Dag bundle, so the
+# Dag processor runs on the same image to parse it.
---
services:
+ airflow-dag-processor:
+ image: airflow-java-worker
+
airflow-worker:
image: airflow-java-worker
volumes:
- ./java-jars:/opt/airflow/java-jars:ro
- ./scala-jars:/opt/airflow/scala-jars:ro
- ./java-test-jars:/opt/airflow/java-test-jars:ro
- command: celery worker -q java,scala,java-test,default
+ command: celery worker -q java,scala,java-test,java-native,default
diff --git a/airflow-e2e-tests/java-native-bundle/.gitignore b/airflow-e2e-tests/java-native-bundle/.gitignore
new file mode 100644
index 00000000000..7f6823bcc0f
--- /dev/null
+++ b/airflow-e2e-tests/java-native-bundle/.gitignore
@@ -0,0 +1,2 @@
+.gradle
+build/
diff --git a/airflow-e2e-tests/java-native-bundle/build.gradle b/airflow-e2e-tests/java-native-bundle/build.gradle
new file mode 100644
index 00000000000..0e572c073fc
--- /dev/null
+++ b/airflow-e2e-tests/java-native-bundle/build.gradle
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+plugins {
+ id("org.apache.airflow.sdk") version "${projectVersion}"
+}
+
+repositories {
+ mavenLocal()
+ mavenCentral()
+}
+
+dependencies {
+ annotationProcessor("org.apache.airflow:airflow-sdk-processor:${projectVersion}")
+ implementation("org.apache.airflow:airflow-sdk:${projectVersion}")
+ implementation("org.apache.airflow:airflow-sdk-jpl:${projectVersion}")
+}
+
+java {
+ toolchain {
+ languageVersion.set(JavaLanguageVersion.of(11))
+ }
+ sourceCompatibility = JavaVersion.VERSION_11
+}
+
+sourceSets {
+ main {
+ java.srcDir("src/java")
+ }
+}
+
+airflowBundle {
+ mainClass = "org.apache.airflow.e2e.NativeBundleBuilder"
+}
diff --git a/airflow-e2e-tests/java-native-bundle/gradle.properties b/airflow-e2e-tests/java-native-bundle/gradle.properties
new file mode 100644
index 00000000000..c3c94e0057f
--- /dev/null
+++ b/airflow-e2e-tests/java-native-bundle/gradle.properties
@@ -0,0 +1,20 @@
+# 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.
+
+org.gradle.configuration-cache=true
+
+projectVersion=1.0.0-SNAPSHOT
diff --git a/airflow-e2e-tests/java-native-bundle/settings.gradle b/airflow-e2e-tests/java-native-bundle/settings.gradle
new file mode 100644
index 00000000000..75016bb20bd
--- /dev/null
+++ b/airflow-e2e-tests/java-native-bundle/settings.gradle
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+// Route the plugin lookup to the SDK build published to the local Maven
+// repository by conftest._setup_java_sdk_integration.
+pluginManagement {
+ repositories {
+ mavenLocal()
+ gradlePluginPortal()
+ }
+}
+
+plugins {
+ id("org.gradle.toolchains.foojay-resolver-convention") version "0.10.0"
+}
+
+rootProject.name = "airflow-e2e-java-native-bundle"
diff --git a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java
new file mode 100644
index 00000000000..17ac91e4cd3
--- /dev/null
+++ b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java
@@ -0,0 +1,84 @@
+/*
+ * 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.
+ */
+
+package org.apache.airflow.e2e;
+
+import java.util.List;
+import org.apache.airflow.e2e.nativedag.AnnotationDag;
+import org.apache.airflow.sdk.*;
+
+/**
+ * Bundle for the native Dag E2E tests: Dags declared entirely in Java, parsed by the Dag processor
+ * from this JAR.
+ *
+ * <p>This is the bundle's main class, so it is the Dag source the Airflow UI shows. Every task sets
+ * {@code queue}, which routes it to the {@code java-native} coordinator.
+ */
+public class NativeBundleBuilder {
+ private static final String QUEUE = "java-native";
+
+ public static class Extract implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ client.setXCom(42L);
+ }
+ }
+
+ public static class Transform implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ var extracted = ((Number) client.getXCom("extract")).longValue();
+ client.setXCom(extracted * 2);
+ }
+ }
+
+ /** Fails the run unless the value reached it, so a successful run proves the XCom flow. */
+ public static class Load implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ var transformed = ((Number) client.getXCom("transform")).longValue();
+ if (transformed != 84L) {
+ throw new IllegalStateException("load expected 84 from transform, got " + transformed);
+ }
+ }
+ }
+
+ public static DagDef buildDag() {
+ var dag =
+ new DagDef("java_native_e2e")
+ .config("description", "Native Java Dag of the Airflow E2E tests")
+ .config("catchup", false)
+ .config("tags", List.of("java-sdk", "native", "e2e"));
+
+ var extract = dag.task("extract", Extract.class).config("queue", QUEUE);
+ var transform = dag.task("transform", Transform.class).config("queue", QUEUE);
+ var load = dag.task("load", Load.class).config("queue", QUEUE);
+
+ transform.after(extract).before(load);
+ return dag;
+ }
+
+ public static Bundle build() {
+ return new Bundle().register(buildDag()).register(AnnotationDag.class);
+ }
+
+ public static void main(String[] args) {
+ Server.create(args).serve(build());
+ }
+}
diff --git a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
new file mode 100644
index 00000000000..e00dccb27b5
--- /dev/null
+++ b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+
+// "native" is a Java keyword, so the native Dags live in "nativedag".
+package org.apache.airflow.e2e.nativedag;
+
+import org.apache.airflow.sdk.*;
+
+/** A native Java Dag declared with annotations; its tasks set {@code queue} like the interface Dag's. */
+@Builder.Dag(
+ id = "java_native_annotation_e2e",
+ description = "Native Java Dag of the Airflow E2E tests, declared with annotations",
+ catchup = false,
+ tags = {"java-sdk", "native", "e2e"})
+public class AnnotationDag {
+ @Builder.Task(id = "extract", queue = "java-native")
+ public long extract() {
+ return 42L;
+ }
+
+ @Builder.Task(id = "transform", queue = "java-native")
+ public long transform(long extracted, double factor) {
+ return (long) (extracted * factor);
+ }
+
+ /** Fails the run unless the value reached it, so a successful run proves the XCom flow. */
+ @Builder.Task(id = "load", queue = "java-native")
+ public void load(long transformed) {
+ if (transformed != 63L) {
+ throw new IllegalStateException("load expected 63 from transform, got " + transformed);
+ }
+ }
+
+ @Builder.Task(id = "audit", queue = "java-native")
+ public void audit() {}
+
+ @Builder.Deps
+ static class Wiring implements AnnotationDagDeps {
+ void depends() {
+ var extracted = extract();
+ load(transform(extracted, lit(1.5)));
+ // Ordering-only edge: audit runs after extract, with no data flowing.
+ extracted.before(audit());
+ }
+ }
+}
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index 3fd6f9a6b6e..aa56af778f8 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -47,6 +47,8 @@ from airflow_e2e_tests.constants import (
GO_SDK_ROOT_PATH,
JAVA_COMPOSE_PATH,
JAVA_DOCKERFILE_PATH,
+ JAVA_NATIVE_BUNDLE_LIBS_PATH,
+ JAVA_NATIVE_BUNDLE_ROOT_PATH,
JAVA_SDK_EXAMPLE_DAGS_PATH,
JAVA_SDK_EXAMPLE_LIBS_PATH,
JAVA_SDK_MAVEN_CACHE_PATH,
@@ -383,8 +385,8 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
console.print("[yellow]Publishing Java SDK artifacts to local Maven repository...")
_run_java_sdk_gradle(JAVA_SDK_ROOT_PATH, "publishToMavenLocal", "-PskipSigning=true", native=native)
- # The example, scala_spark_example, and java-test-bundle are independent
- # Gradle builds that all consume the SDK artifact published above, so build
+ # The example, scala_spark_example, java-test-bundle, and java-native-bundle
+ # are independent Gradle builds that all consume the SDK artifact published above, so build
# them concurrently. Sharing a writable Gradle user home between concurrent
# builds is safe because each build can ping the other's lock-owner port over
# one shared loopback - the host's own in native mode, --network=host in the
@@ -399,14 +401,17 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
rmtree(JAVA_SDK_EXAMPLE_LIBS_PATH, ignore_errors=True)
rmtree(SCALA_SPARK_EXAMPLE_LIBS_PATH, ignore_errors=True)
rmtree(JAVA_TEST_BUNDLE_LIBS_PATH, ignore_errors=True)
+ rmtree(JAVA_NATIVE_BUNDLE_LIBS_PATH, ignore_errors=True)
toolchain = "host toolchain" if native else "eclipse-temurin:17-jdk"
console.print(
- f"[yellow]Building Java SDK, Scala Spark, and test-fixture bundles concurrently ({toolchain})..."
+ "[yellow]Building Java SDK, Scala Spark, test-fixture, and native-Dag bundles concurrently "
+ f"({toolchain})..."
)
example_bundle_workdirs = [
JAVA_SDK_ROOT_PATH / "example",
JAVA_SDK_ROOT_PATH / "scala_spark_example",
JAVA_TEST_BUNDLE_ROOT_PATH,
+ JAVA_NATIVE_BUNDLE_ROOT_PATH,
]
with ThreadPoolExecutor(max_workers=len(example_bundle_workdirs)) as pool:
bundle_builds = [
@@ -424,6 +429,8 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
copytree(JAVA_SDK_EXAMPLE_LIBS_PATH, tmp_dir / "java-jars")
copytree(SCALA_SPARK_EXAMPLE_LIBS_PATH, tmp_dir / "scala-jars")
copytree(JAVA_TEST_BUNDLE_LIBS_PATH, tmp_dir / "java-test-jars")
+ # The native-Dag bundle goes into the Dag bundle, where the Dag processor parses it.
+ copytree(JAVA_NATIVE_BUNDLE_LIBS_PATH, tmp_dir / "dags" / "java-native")
# Copy the Java SDK example Dag files so Airflow can discover them.
copyfile(JAVA_SDK_EXAMPLE_DAGS_PATH / "java_examples.py", tmp_dir / "dags" / "java_examples.py")
@@ -437,7 +444,7 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
# JRE and copies nothing from the context, so without this docker build would
# tar and stream the bundles (hundreds of MB of Spark JARs) to the daemon for
# nothing. The JARs reach the worker via the compose bind-mounts, not the image.
- (tmp_dir / ".dockerignore").write_text("java-jars/\nscala-jars/\njava-test-jars/\n")
+ (tmp_dir / ".dockerignore").write_text("java-jars/\nscala-jars/\njava-test-jars/\ndags/\n")
# Build a local Docker image that extends DOCKER_IMAGE with a JRE.
# We do this explicitly so testcontainers' DockerCompose.start() does not
@@ -483,10 +490,16 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir):
"classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
"kwargs": {"jars_root": ["/opt/airflow/java-test-jars"]},
},
+ # No root, so it serves the Dag bundle: it parses the native-Dag JAR
+ # there and runs that JAR's tasks.
+ "java-native": {
+ "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
+ "kwargs": {},
+ },
}
)
queue_to_coordinator = json.dumps(
- {"java": "java-jdk", "scala": "scala-jdk", "java-test": "java-test-jdk"}
+ {"java": "java-jdk", "scala": "scala-jdk", "java-test": "java-test-jdk", "java-native": "java-native"}
)
# Connection expected by the Java example bundle tasks. The JSON form
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
index 4a1a2f7be38..a9d1eab2c77 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
@@ -84,6 +84,11 @@ JAVA_TEST_BUNDLE_ROOT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "java-tes
JAVA_TEST_BUNDLE_DAGS_PATH = JAVA_TEST_BUNDLE_ROOT_PATH / "src" / "resources" / "dags"
JAVA_TEST_BUNDLE_LIBS_PATH = JAVA_TEST_BUNDLE_ROOT_PATH / "build" / "bundle"
+# Java native-Dag bundle paths (Dags declared in Java, parsed from the JAR by the
+# Dag processor; its JAR goes into the Dag bundle rather than a coordinator root).
+JAVA_NATIVE_BUNDLE_ROOT_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "java-native-bundle"
+JAVA_NATIVE_BUNDLE_LIBS_PATH = JAVA_NATIVE_BUNDLE_ROOT_PATH / "build" / "bundle"
+
# Go SDK E2E test paths
GO_SDK_ROOT_PATH = AIRFLOW_ROOT_PATH / "go-sdk"
GO_SDK_DAGS_PATH = GO_SDK_ROOT_PATH / "dags"
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py
new file mode 100644
index 00000000000..f8c00b31fb3
--- /dev/null
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py
@@ -0,0 +1,153 @@
+# 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.
+"""
+End-to-end test of Dags declared entirely in Java.
+
+Run with::
+
+ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \\
+ tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py -xvs
+
+No Python file declares these Dags. The ``airflow-e2e-tests/java-native-bundle`` JAR sits in the Dag
+bundle, where the root-less ``java-native`` coordinator claims it: the Dag processor runs the JAR to
+parse it, and the worker runs the same JAR for each task, which every task routes to with its
+``queue``. ``java_native_e2e`` is declared with the interface API, ``java_native_annotation_e2e``
+with annotations.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+
+import pytest
+
+from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
+
+# The Dag processor starts a JVM to parse the JAR, and each task starts another.
+_JAVA_TASK_TIMEOUT = 600
+
+_QUEUE = "java-native"
+
+
+@dataclass(frozen=True)
+class _NativeDag:
+ dag_id: str
+ downstream: dict[str, set[str]]
+ return_values: dict[str, int]
+
+
+_NATIVE_DAGS = [
+ _NativeDag(
+ dag_id="java_native_e2e",
+ downstream={"extract": {"transform"}, "transform": {"load"}, "load": set()},
+ return_values={"extract": 42, "transform": 84},
+ ),
+ _NativeDag(
+ dag_id="java_native_annotation_e2e",
+ downstream={"extract": {"transform", "audit"}, "transform": {"load"}, "load": set(), "audit": set()},
+ # transform(extracted, lit(1.5)).
+ return_values={"extract": 42, "transform": 63},
+ ),
+]
+
+_by_dag_id = pytest.mark.parametrize("native_dag", _NATIVE_DAGS, ids=lambda d: d.dag_id)
+
+
+@dataclass
+class _CompletedRun:
+ run_id: str
+ state: str
+ ti_states: dict[str, str]
+
+
+@pytest.fixture(scope="module")
+def parsed_dags() -> AirflowClient:
+ """A client that has waited for the Dag processor to register both Dags from the JAR."""
+ client = AirflowClient()
+ for native_dag in _NATIVE_DAGS:
+ client.wait_for_dag(native_dag.dag_id, timeout=_JAVA_TASK_TIMEOUT)
+ return client
+
+
+@pytest.fixture(scope="module")
+def completed_runs(parsed_dags: AirflowClient) -> dict[str, _CompletedRun]:
+ """Trigger both Dags at once, then wait for both runs."""
+ client = parsed_dags
+ run_ids = {
+ native_dag.dag_id: client.trigger_dag(
+ native_dag.dag_id, json={"logical_date": datetime.now(timezone.utc).isoformat()}
+ )["dag_run_id"]
+ for native_dag in _NATIVE_DAGS
+ }
+ runs = {}
+ for dag_id, run_id in run_ids.items():
+ state = client.wait_for_dag_run(dag_id=dag_id, run_id=run_id, timeout=_JAVA_TASK_TIMEOUT)
+ ti_resp = client.get_task_instances(dag_id=dag_id, run_id=run_id)
+ runs[dag_id] = _CompletedRun(
+ run_id=run_id,
+ state=state,
+ ti_states={ti["task_id"]: ti.get("state") for ti in ti_resp.get("task_instances", [])},
+ )
+ return runs
+
+
+@_by_dag_id
+def test_the_graph_is_the_one_java_declared(parsed_dags: AirflowClient, native_dag: _NativeDag):
+ tasks = parsed_dags.get_tasks(native_dag.dag_id).get("tasks", [])
+
+ assert {task["task_id"]: set(task["downstream_task_ids"]) for task in tasks} == native_dag.downstream
+
+
+@_by_dag_id
+def test_every_task_is_routed_to_the_native_coordinator(parsed_dags: AirflowClient, native_dag: _NativeDag):
+ """The Java DSL has no Dag-level queue, so each task sets its own."""
+ tasks = parsed_dags.get_tasks(native_dag.dag_id).get("tasks", [])
+
+ assert {task["task_id"]: task["queue"] for task in tasks} == dict.fromkeys(native_dag.downstream, _QUEUE)
+
+
+@_by_dag_id
+def test_the_dag_source_is_the_bundle_main_class(parsed_dags: AirflowClient, native_dag: _NativeDag):
+ """The Code view shows the Java source the JAR embeds, not the JAR read as text."""
+ content = parsed_dags.get_dag_source(native_dag.dag_id)["content"]
+
+ assert "public class NativeBundleBuilder" in content
+ assert 'new DagDef("java_native_e2e")' in content
+
+
+@_by_dag_id
+def test_dag_run_succeeded(completed_runs: dict[str, _CompletedRun], native_dag: _NativeDag):
+ run = completed_runs[native_dag.dag_id]
+
+ assert run.state == "success", (
+ f"expected the run to succeed; got {run.state!r}. task states: {run.ti_states}"
+ )
+ # load throws unless transform's value reached it, so its success proves the last hop.
+ assert run.ti_states == dict.fromkeys(native_dag.downstream, "success")
+
+
+@_by_dag_id
+def test_xcoms_flow_between_java_tasks(
+ parsed_dags: AirflowClient, completed_runs: dict[str, _CompletedRun], native_dag: _NativeDag
+):
+ run_id = completed_runs[native_dag.dag_id].run_id
+ for task_id, expected in native_dag.return_values.items():
+ value = parsed_dags.get_xcom_value(
+ dag_id=native_dag.dag_id, task_id=task_id, run_id=run_id, key="return_value"
+ ).get("value")
+ assert value == expected, f"{native_dag.dag_id}.{task_id} returned {value!r}"
(airflow) 02/08: Parse the native TypeScript Dag in the compose e2e
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 75ac609e9bc561d97820dbf13957d549a97e4790
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 07:44:42 2026 +0000
Parse the native TypeScript Dag in the compose e2e
Put the example bundle in the Dag bundle as well, add a root-less
"ts-native" Node coordinator that claims it there, and give the Dag
processor node. The existing "ts" coordinator still runs every TypeScript
task, so the native Dag test needs no gate anymore.
---
airflow-e2e-tests/docker/ts.yml | 13 +++++++++++--
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py | 15 ++++++++++++++-
.../ts_sdk_tests/test_ts_sdk_native_dag.py | 17 +++--------------
3 files changed, 28 insertions(+), 17 deletions(-)
diff --git a/airflow-e2e-tests/docker/ts.yml b/airflow-e2e-tests/docker/ts.yml
index 056106ff9d5..902ce7562c9 100644
--- a/airflow-e2e-tests/docker/ts.yml
+++ b/airflow-e2e-tests/docker/ts.yml
@@ -17,10 +17,12 @@
# Docker Compose override for ts_sdk E2E test mode.
#
-# The stock worker image ships no Node.js runtime, so node-provider copies the
+# The stock Airflow image ships no Node.js runtime, so node-provider copies the
# node binary from the same image conftest builds the bundle with into a
# shared volume. The bundle is bind-mounted where NodeCoordinator scans, and
-# the worker consumes the "typescript" queue where @task.stub tasks are routed.
+# the worker consumes the "typescript" queue where TypeScript tasks are routed.
+# The Dag processor gets node too, to parse the bundle's native Dag from the
+# Dag bundle.
---
services:
node-provider:
@@ -29,6 +31,13 @@ services:
volumes:
- nodejs-bin:/opt/nodejs
+ airflow-dag-processor:
+ volumes:
+ - nodejs-bin:/opt/nodejs:ro
+ depends_on:
+ node-provider:
+ condition: service_completed_successfully
+
airflow-worker:
volumes:
- ./ts-bundles:/opt/airflow/ts-bundles:ro
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index 39ff3fd0a64..4b441bb549f 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -724,12 +724,21 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir):
ts_bundles_dir.mkdir()
# Deliberately renamed: the coordinator routes on embedded metadata, not on a fixed name.
copyfile(TS_SDK_EXAMPLE_PATH / "dist" / "bundle.min.mjs", ts_bundles_dir / "example.min.mjs")
+ # The same artifact inside the Dag bundle, where the Dag processor finds its native Dag.
+ (tmp_dir / "dags" / "typescript").mkdir()
+ copyfile(
+ TS_SDK_EXAMPLE_PATH / "dist" / "bundle.min.mjs", tmp_dir / "dags" / "typescript" / "example.min.mjs"
+ )
# Both of the example bundle's Dags: one bundle.mjs provides for two dag_ids,
# and the tests check that dispatch tells their same-named tasks apart.
for dag_file in ("typescript_example.py", "typescript_taskflow_example.py"):
copyfile(TS_SDK_EXAMPLE_PATH / "dags" / dag_file, tmp_dir / "dags" / dag_file)
+ # "ts" runs every TypeScript task, stub or native, from /opt/airflow/ts-bundles.
+ # "ts-native" has no root, so it serves the Dag bundle: it parses the bundle's
+ # native Dag, in the Dag processor and on the worker that runs the Dag's
+ # Python task (trigger_downstream).
coordinator_config = json.dumps(
{
"ts": {
@@ -738,7 +747,11 @@ def _setup_ts_sdk_integration(dot_env_file, tmp_dir):
"bundles_root": ["/opt/airflow/ts-bundles"],
"node_executable": "/opt/nodejs/node",
},
- }
+ },
+ "ts-native": {
+ "classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
+ "kwargs": {"node_executable": "/opt/nodejs/node"},
+ },
}
)
queue_to_coordinator = json.dumps({"typescript": "ts"})
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
index f23684d7a5b..f8aa1f48a72 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
@@ -19,8 +19,7 @@ End-to-end test of a Dag declared entirely in TypeScript.
Run with::
- E2E_TEST_MODE=ts_sdk RUN_TS_SDK_NATIVE_DAG_TESTS=true \\
- uv run --project airflow-e2e-tests pytest \\
+ E2E_TEST_MODE=ts_sdk uv run --project airflow-e2e-tests pytest \\
tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py -xvs
Unlike ``test_ts_sdk_dag.py``, no Python file declares this Dag: the Dag processor asks the
@@ -29,14 +28,12 @@ Unlike ``test_ts_sdk_dag.py``, no Python file declares this Dag: the Dag process
named fan-in, order-only edges, a conditional, a multi-way branch, and a ``TriggerDagRunOperator`` a
Python worker runs.
-Gated behind ``RUN_TS_SDK_NATIVE_DAG_TESTS`` because it needs a Dag processor that can dispatch a
-parse request to a language coordinator. Until that lands, the bundle answers a request nothing
-sends, and the Dag never appears.
+The bundle sits in the Dag bundle, where the ``ts-native`` coordinator claims it. The Python worker
+that runs ``trigger_downstream`` parses the bundle the same way to rebuild the operator.
"""
from __future__ import annotations
-import os
from dataclasses import dataclass
from datetime import datetime, timezone
@@ -44,8 +41,6 @@ import pytest
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
-_RUN_NATIVE = os.environ.get("RUN_TS_SDK_NATIVE_DAG_TESTS", "").lower() in ("true", "1")
-
# Parsing the bundle launches node before the first task is even scheduled, so
# allow the same headroom the mixed-language suite does.
_TS_TASK_TIMEOUT = 600
@@ -57,12 +52,6 @@ _NORTH_ROWS_VARIABLE = "typescript_native_north_rows"
_SOUTH_ROWS_VARIABLE = "typescript_native_south_rows"
_CADENCE_VARIABLE = "typescript_native_cadence"
-pytestmark = pytest.mark.skipif(
- not _RUN_NATIVE,
- reason="Needs a Dag processor that dispatches parse requests to a language coordinator "
- "(RUN_TS_SDK_NATIVE_DAG_TESTS)",
-)
-
@dataclass
class _CompletedRun:
(airflow) 04/08: Keep the containerized Java build's Maven repository on the host
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6707aa83b8c998f416858c50ab0fbb05b2a21ff5
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 08:01:29 2026 +0000
Keep the containerized Java build's Maven repository on the host
The JVM takes user.home from /etc/passwd, so when the host UID is one the
JDK image defines (1000 is its "ubuntu" user), publishToMavenLocal wrote
into the container and the example builds could not find the SDK. Name the
mounted repository with maven.repo.local instead of relying on HOME.
---
airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index 4b441bb549f..3fd6f9a6b6e 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -319,6 +319,9 @@ def _run_java_sdk_gradle(workdir, *gradle_argv, capture_output=False, native=Fal
image's HOME (/root) which the non-root process cannot write to.
* files/m2 is mounted directly as ~/.m2 so publishToMavenLocal writes
there without nesting, and its contents are visible on the host.
+ maven.repo.local names it explicitly: the JVM takes its home from
+ /etc/passwd, not HOME, so a host UID that the image does define (such
+ as 1000, its "ubuntu" user) would otherwise publish into the container.
"""
if native:
cwd = workdir
@@ -351,6 +354,7 @@ def _run_java_sdk_gradle(workdir, *gradle_argv, capture_output=False, native=Fal
"eclipse-temurin:17-jdk",
"/repo/java-sdk/gradlew",
"--no-daemon",
+ "-Dmaven.repo.local=/workspace-home/.m2/repository",
*gradle_argv,
]
return subprocess.run(argv, cwd=cwd, check=True, capture_output=capture_output, text=True)
(airflow) 06/08: Run the Lang-SDK e2e jobs when native Dag parsing changes
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9e4160d7169b5323a54c4c238f44e90310d59aa1
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 08:02:53 2026 +0000
Run the Lang-SDK e2e jobs when native Dag parsing changes
The Java and TypeScript e2e suites now parse native Dags through the Dag
processor, the Dag bag and the coordinators' Dag importers, so changes to
those sources select both jobs. The Java native-Dag bundle selects the
Java job.
---
dev/breeze/doc/ci/04_selective_checks.md | 16 ++++++---
.../src/airflow_breeze/utils/selective_checks.py | 13 ++++++++
dev/breeze/tests/test_selective_checks.py | 38 ++++++++++++++++++++++
3 files changed, 62 insertions(+), 5 deletions(-)
diff --git a/dev/breeze/doc/ci/04_selective_checks.md b/dev/breeze/doc/ci/04_selective_checks.md
index 2c3f81b506e..fa7c675a4e6 100644
--- a/dev/breeze/doc/ci/04_selective_checks.md
+++ b/dev/breeze/doc/ci/04_selective_checks.md
@@ -446,11 +446,17 @@ together using `pytest-xdist` (pytest-xdist distributes the tests among parallel
or tests), then tests for all Providers are run
* `Java SDK E2E tests` (the `java_sdk` mode of the deployed-stack tests, exposed as the
`run-java-sdk-e2e-tests` output) run when the Java SDK sources (`java-sdk/`, excluding `.md`), the
- Java test-fixture bundle (`airflow-e2e-tests/java-test-bundle/`), the Java e2e suite or its Docker
+ Java test-fixture and native-Dag bundles (`airflow-e2e-tests/java-test-bundle/`,
+ `airflow-e2e-tests/java-native-bundle/`), the Java e2e suite or its Docker
files (`airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/`,
- `airflow-e2e-tests/docker/java.yml`, `airflow-e2e-tests/docker/Dockerfile.java`), or the Java
- coordinator (`task-sdk/src/airflow/sdk/coordinators/java/`, `_subprocess.py`) change. Like the
- other deployed e2e suites, enabling them forces `PROD Image building`.
+ `airflow-e2e-tests/docker/java.yml`, `airflow-e2e-tests/docker/Dockerfile.java`), the Java
+ coordinator (`task-sdk/src/airflow/sdk/coordinators/java/`, `_subprocess.py`), or the native
+ Lang-SDK Dag parsing sources change. Those are the Dag processing files `dagbag.py`,
+ `importer_routing.py`, `manager.py` and `processor.py`, `models/dagcode.py`, the Task SDK importers
+ (`task-sdk/src/airflow/sdk/importers/`), the coordinator registry
+ (`task-sdk/src/airflow/sdk/execution_time/coordinator.py`), and `coordinators/_dag_importer.py`
+ and `coordinators/_materialize.py`; they trigger the TypeScript SDK E2E tests too. Like the other
+ deployed e2e suites, enabling them forces `PROD Image building`.
* `OpenLineage E2E tests` (the `openlineage` mode of the deployed-stack tests under
`airflow-e2e-tests/tests/airflow_e2e_tests/openlineage_tests`, exposed as the
`run-openlineage-e2e-tests` output) run when the `openlineage` or `common` providers or the
@@ -610,7 +616,7 @@ GitHub Actions to pass the list of parameters to a command to execute
| run-system-tests | Whether system tests should be run ("true"/"false") | true | |
| run-task-sdk-tests | Whether Task SDK tests should be run ("true"/"false") | true | |
| run-ts-sdk-docs | Whether the TypeScript SDK API reference should be built — on `ts-sdk/api-docs/`, `ts-sdk/docs/`, or `ts-sdk/src/` changes, including Markdown ("true"/"false") | true | |
-| run-ts-sdk-e2e-tests | Whether TypeScript SDK e2e tests should be run — on runtime-affecting `ts-sdk/`, TS e2e test, or Node coordinator changes ("true"/"false") | true | |
+| run-ts-sdk-e2e-tests | Whether TypeScript SDK e2e tests should be run — on runtime-affecting `ts-sdk/`, TS e2e test, Node coordinator, or native Lang-SDK Dag parsing changes ("true"/"false") | true | |
| run-ui-tests | Whether UI tests should be run ("true"/"false") | true | |
| run-unit-tests | Whether unit tests should be run ("true"/"false") | true | |
| run-www-tests | Whether Legacy WWW tests should be run ("true"/"false") | true | |
diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
index 842d3a017c9..b95ab843fa2 100644
--- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
+++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
@@ -180,6 +180,16 @@ class HashableDict(dict[T, list[str]]):
return hash(frozenset(self))
+# Core and Task SDK sources on the native Lang-SDK Dag path: discovery, parsing through a
+# coordinator's runtime, rebuilding the Dag for a Dag bag, and reading its source.
+LANG_SDK_NATIVE_DAG_PARSING_FILES = (
+ r"^airflow-core/src/airflow/dag_processing/(dagbag|importer_routing|manager|processor)\.py$",
+ r"^airflow-core/src/airflow/models/dagcode\.py$",
+ r"^task-sdk/src/airflow/sdk/coordinators/_(dag_importer|materialize)\.py$",
+ r"^task-sdk/src/airflow/sdk/execution_time/coordinator\.py$",
+ r"^task-sdk/src/airflow/sdk/importers/.*",
+)
+
CI_FILE_GROUP_MATCHES: HashableDict[FileGroupForCi] = HashableDict(
{
FileGroupForCi.ENVIRONMENT_FILES: [
@@ -246,11 +256,13 @@ CI_FILE_GROUP_MATCHES: HashableDict[FileGroupForCi] = HashableDict(
# `.md` excluded — doc-only edits do not affect the Gradle build.
r"^java-sdk/(?!.*\.md$).*",
r"^airflow-e2e-tests/java-test-bundle/.*",
+ r"^airflow-e2e-tests/java-native-bundle/.*",
r"^airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/.*",
r"^airflow-e2e-tests/docker/java\.yml$",
r"^airflow-e2e-tests/docker/Dockerfile\.java$",
r"^task-sdk/src/airflow/sdk/coordinators/_subprocess\.py$",
r"^task-sdk/src/airflow/sdk/coordinators/java/.*",
+ *LANG_SDK_NATIVE_DAG_PARSING_FILES,
],
FileGroupForCi.GO_SDK_E2E_FILES: [
# `.md` excluded — doc-only edits do not affect the Go build or e2e tests.
@@ -283,6 +295,7 @@ CI_FILE_GROUP_MATCHES: HashableDict[FileGroupForCi] = HashableDict(
r"^airflow-e2e-tests/docker/ts\.yml$",
r"^task-sdk/src/airflow/sdk/coordinators/_subprocess\.py$",
r"^task-sdk/src/airflow/sdk/coordinators/node/.*",
+ *LANG_SDK_NATIVE_DAG_PARSING_FILES,
],
FileGroupForCi.PYTHON_PRODUCTION_FILES: [
# Production Python source the runtime ships — excludes tests, docs,
diff --git a/dev/breeze/tests/test_selective_checks.py b/dev/breeze/tests/test_selective_checks.py
index be57fd40b15..72327a6348b 100644
--- a/dev/breeze/tests/test_selective_checks.py
+++ b/dev/breeze/tests/test_selective_checks.py
@@ -1560,6 +1560,44 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
},
id="Run java e2e tests when JavaCoordinator changes",
),
+ pytest.param(
+ ("task-sdk/src/airflow/sdk/coordinators/_dag_importer.py",),
+ {
+ "run-java-sdk-e2e-tests": "true",
+ "run-ts-sdk-e2e-tests": "true",
+ "prod-image-build": "true",
+ },
+ id="Run java and ts e2e tests when the coordinator Dag importer changes",
+ ),
+ pytest.param(
+ ("airflow-core/src/airflow/dag_processing/manager.py",),
+ {
+ "run-java-sdk-e2e-tests": "true",
+ "run-ts-sdk-e2e-tests": "true",
+ "prod-image-build": "true",
+ },
+ id="Run java and ts e2e tests when the Dag processor manager changes",
+ ),
+ pytest.param(
+ (
+ "airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java",
+ ),
+ {
+ "run-java-sdk-tests": "false",
+ "run-java-sdk-e2e-tests": "true",
+ "run-ts-sdk-e2e-tests": "false",
+ "prod-image-build": "true",
+ },
+ id="Run only java e2e tests when the java native-Dag bundle changes",
+ ),
+ pytest.param(
+ ("airflow-core/src/airflow/dag_processing/bundles/base.py",),
+ {
+ "run-java-sdk-e2e-tests": "false",
+ "run-ts-sdk-e2e-tests": "false",
+ },
+ id="Skip java and ts e2e tests for a Dag processing change off the native Dag path",
+ ),
pytest.param(
("go-sdk/sdk/variable.go",),
{
(airflow) 07/08: Share the Dag run lookup and the native Java queue name
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 9582bdaaf284e1528a699f599965e13a4ac1921a
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 09:02:43 2026 +0000
Share the Dag run lookup and the native Java queue name
wait_for_dag_run now polls through get_dag_run, and the annotation Dag
takes its queue from the interface Dag's constant.
---
.../src/java/org/apache/airflow/e2e/NativeBundleBuilder.java | 2 +-
.../java/org/apache/airflow/e2e/nativedag/AnnotationDag.java | 12 +++++++-----
.../tests/airflow_e2e_tests/e2e_test_utils/clients.py | 7 ++-----
.../java_sdk_tests/test_java_sdk_native_dag.py | 7 ++++++-
4 files changed, 16 insertions(+), 12 deletions(-)
diff --git a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java
index 17ac91e4cd3..cfddd691de2 100644
--- a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java
+++ b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/NativeBundleBuilder.java
@@ -31,7 +31,7 @@ import org.apache.airflow.sdk.*;
* {@code queue}, which routes it to the {@code java-native} coordinator.
*/
public class NativeBundleBuilder {
- private static final String QUEUE = "java-native";
+ public static final String QUEUE = "java-native";
public static class Extract implements Task {
@Override
diff --git a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
index e00dccb27b5..89abab7737b 100644
--- a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
+++ b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/AnnotationDag.java
@@ -20,34 +20,36 @@
// "native" is a Java keyword, so the native Dags live in "nativedag".
package org.apache.airflow.e2e.nativedag;
+import static org.apache.airflow.e2e.NativeBundleBuilder.QUEUE;
+
import org.apache.airflow.sdk.*;
-/** A native Java Dag declared with annotations; its tasks set {@code queue} like the interface Dag's. */
+/** A native Java Dag declared with annotations; its tasks use the interface Dag's queue. */
@Builder.Dag(
id = "java_native_annotation_e2e",
description = "Native Java Dag of the Airflow E2E tests, declared with annotations",
catchup = false,
tags = {"java-sdk", "native", "e2e"})
public class AnnotationDag {
- @Builder.Task(id = "extract", queue = "java-native")
+ @Builder.Task(id = "extract", queue = QUEUE)
public long extract() {
return 42L;
}
- @Builder.Task(id = "transform", queue = "java-native")
+ @Builder.Task(id = "transform", queue = QUEUE)
public long transform(long extracted, double factor) {
return (long) (extracted * factor);
}
/** Fails the run unless the value reached it, so a successful run proves the XCom flow. */
- @Builder.Task(id = "load", queue = "java-native")
+ @Builder.Task(id = "load", queue = QUEUE)
public void load(long transformed) {
if (transformed != 63L) {
throw new IllegalStateException("load expected 63 from transform, got " + transformed);
}
}
- @Builder.Task(id = "audit", queue = "java-native")
+ @Builder.Task(id = "audit", queue = QUEUE)
public void audit() {}
@Builder.Deps
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
index ee541f97f7d..bd80556b586 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
@@ -144,11 +144,7 @@ class AirflowClient:
def wait_for_dag_run(self, dag_id: str, run_id: str, timeout=300, check_interval=5):
start_time = time.time()
while time.time() - start_time < timeout:
- response = self._make_request(
- method="GET",
- endpoint=f"dags/{dag_id}/dagRuns/{run_id}",
- )
- state = response.get("state")
+ state = self.get_dag_run(dag_id=dag_id, run_id=run_id).get("state")
if state in {"success", "failed"}:
return state
time.sleep(check_interval)
@@ -184,6 +180,7 @@ class AirflowClient:
return self._make_request(method="GET", endpoint=f"dagSources/{dag_id}")
def get_dag_run(self, dag_id: str, run_id: str):
+ """Get a Dag run, with its state, run type and conf."""
return self._make_request(method="GET", endpoint=f"dags/{dag_id}/dagRuns/{run_id}")
def trigger_dag_and_wait(self, dag_id: str, json=None):
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py
index f8c00b31fb3..f7296c1f96b 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_native_dag.py
@@ -123,7 +123,12 @@ def test_every_task_is_routed_to_the_native_coordinator(parsed_dags: AirflowClie
@_by_dag_id
def test_the_dag_source_is_the_bundle_main_class(parsed_dags: AirflowClient, native_dag: _NativeDag):
- """The Code view shows the Java source the JAR embeds, not the JAR read as text."""
+ """
+ The Code view shows the Java source the JAR embeds, not the JAR read as text.
+
+ A JAR embeds one source, its main class, so both Dags show the file that declares
+ ``java_native_e2e`` and registers ``java_native_annotation_e2e``.
+ """
content = parsed_dags.get_dag_source(native_dag.dag_id)["content"]
assert "public class NativeBundleBuilder" in content
(airflow) 08/08: Build the k8s TypeScript artifacts only for the native Dag test
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit dedaa3048e3b5b3d382cf160beb9544e1d4ba221
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 12:34:28 2026 +0000
Build the k8s TypeScript artifacts only for the native Dag test
setup-lang-sdk-test packed the TypeScript bundle and built its worker image
for every lang-SDK run, so the Go and Java k8s job, which provisions no Node
toolchain, failed during setup. Build and upload them only with
--ts-sdk-native-dag-test (RUN_TS_SDK_NATIVE_DAG_K8S_TESTS), the variable the
native TypeScript k8s test is gated on.
---
dev/breeze/doc/images/output_k8s.svg | 26 ++---
dev/breeze/doc/images/output_k8s.txt | 2 +-
.../doc/images/output_k8s_setup-lang-sdk-test.svg | 96 +++++++++++++-----
.../doc/images/output_k8s_setup-lang-sdk-test.txt | 2 +-
.../airflow_breeze/commands/kubernetes_commands.py | 100 ++++++++++++-------
.../commands/kubernetes_commands_config.py | 1 +
.../tests/test_kubernetes_lang_sdk_commands.py | 108 +++++++++++++++++++++
7 files changed, 261 insertions(+), 74 deletions(-)
diff --git a/dev/breeze/doc/images/output_k8s.svg b/dev/breeze/doc/images/output_k8s.svg
index 92f835a53e8..8c3c725e05c 100644
--- a/dev/breeze/doc/images/output_k8s.svg
+++ b/dev/breeze/doc/images/output_k8s.svg
@@ -1,4 +1,4 @@
-<svg class="rich-terminal" viewBox="0 0 1482 1026.0" xmlns="http://www.w3.org/2000/svg">
+<svg class="rich-terminal" viewBox="0 0 1482 1050.4" xmlns="http://www.w3.org/2000/svg">
<!-- Generated with Rich https://www.textualize.io -->
<style>
@@ -43,7 +43,7 @@
<defs>
<clipPath id="breeze-k8s-clip-terminal">
- <rect x="0" y="0" width="1463.0" height="975.0" />
+ <rect x="0" y="0" width="1463.0" height="999.4" />
</clipPath>
<clipPath id="breeze-k8s-line-0">
<rect x="0" y="1.5" width="1464" height="24.65"/>
@@ -162,9 +162,12 @@
<clipPath id="breeze-k8s-line-38">
<rect x="0" y="928.7" width="1464" height="24.65"/>
</clipPath>
+<clipPath id="breeze-k8s-line-39">
+ <rect x="0" y="953.1" width="1464" height="24.65"/>
+ </clipPath>
</defs>
- <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="1024" rx="8"/><text class="breeze-k8s-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: k8s</text>
+ <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="1048.4" rx="8"/><text class="breeze-k8s-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: k8s</text>
<g transform="translate(26,22)">
<circle cx="0" cy="0" r="7" fill="#ff5f57"/>
<circle cx="22" cy="0" r="7" fill="#febc2e"/>
@@ -206,14 +209,15 @@
</text><text class="breeze-k8s-r5" x="0" y="727.6" textLength="12.2" clip-path="url(#breeze-k8s-line-29)">│</text><text class="breeze-k8s-r1" x="280.6" y="727.6" textLength="1159" clip-path="url(#breeze-k8s-line-29)">resources, and run the optional per-overlay pytest module.                              &# [...]
</text><text class="breeze-k8s-r5" x="0" y="752" textLength="12.2" clip-path="url(#breeze-k8s-line-30)">│</text><text class="breeze-k8s-r4" x="24.4" y="752" textLength="231.8" clip-path="url(#breeze-k8s-line-30)">run-complete-tests </text><text class="breeze-k8s-r1" x="280.6" y="752" textLength="1159" clip-path="url(#breeze-k8s-line-30)">Run complete k8s tests consisting of: creating cluster, building and uploading image, d [...]
</text><text class="breeze-k8s-r5" x="0" y="776.4" textLength="12.2" clip-path="url(#breeze-k8s-line-31)">│</text><text class="breeze-k8s-r1" x="280.6" y="776.4" textLength="1159" clip-path="url(#breeze-k8s-line-31)">airflow, running tests and deleting clusters (optionally for all clusters in parallel).        </text><text class="breeze-k8s-r5" x="1451.8" y="776.4" textLength="12.2" clip-path=" [...]
-</text><text class="breeze-k8s-r5" x="0" y="800.8" textLength="12.2" clip-path="url(#breeze-k8s-line-32)">│</text><text class="breeze-k8s-r4" x="24.4" y="800.8" textLength="231.8" clip-path="url(#breeze-k8s-line-32)">setup-lang-sdk-test</text><text class="breeze-k8s-r1" x="280.6" y="800.8" textLength="1159" clip-path="url(#breeze-k8s-line-32)">Provision the lang-SDK (Go + Java + TypeScript) coordinator system test on an alr [...]
-</text><text class="breeze-k8s-r5" x="0" y="825.2" textLength="12.2" clip-path="url(#breeze-k8s-line-33)">│</text><text class="breeze-k8s-r1" x="280.6" y="825.2" textLength="1159" clip-path="url(#breeze-k8s-line-33)">KubernetesExecutor cluster: build artifacts, build + load the Java and TypeScript worker       </text><text class="breeze-k8s-r5" x="1451.8" y="825.2" textLength="12.2" clip-path="url(# [...]
-</text><text class="breeze-k8s-r5" x="0" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-line-34)">│</text><text class="breeze-k8s-r1" x="280.6" y="849.6" textLength="1159" clip-path="url(#breeze-k8s-line-34)">images, deploy localstack S3, upload artifacts + stub Dag, create config, and upgrade the Helm </text><text class="breeze-k8s-r5" x="1451.8" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-line [...]
-</text><text class="breeze-k8s-r5" x="0" y="874" textLength="12.2" clip-path="url(#breeze-k8s-line-35)">│</text><text class="breeze-k8s-r1" x="280.6" y="874" textLength="1024.8" clip-path="url(#breeze-k8s-line-35)">release. Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests </text><text class="breeze-k8s-r4" x="1305.4" y="874" textLength="122" clip-path="url(#breeze-k8s-line-35)">--executor</text><text class="breez [...]
-</text><text class="breeze-k8s-r5" x="0" y="898.4" textLength="12.2" clip-path="url(#breeze-k8s-line-36)">│</text><text class="breeze-k8s-r1" x="280.6" y="898.4" textLength="268.4" clip-path="url(#breeze-k8s-line-36)">KubernetesExecutor -- </text><text class="breeze-k8s-r6" x="549" y="898.4" textLength="24.4" clip-path="url(#breeze-k8s-line-36)">-k</text><text class="breeze-k8s-r1" x="573.4" y="898.4" textLength="866.2" clip-path="url(#breeze-k8s-line-36)"> test_lang_sdk_c [...]
-</text><text class="breeze-k8s-r5" x="0" y="922.8" textLength="12.2" clip-path="url(#breeze-k8s-line-37)">│</text><text class="breeze-k8s-r4" x="24.4" y="922.8" textLength="231.8" clip-path="url(#breeze-k8s-line-37)">shell              </text><text class="breeze-k8s-r1" x="280.6" y="922.8" textLength="1159" clip-path="url(#breeze-k8s-line-37)">Run shell environment for the current KinD [...]
-</text><text class="breeze-k8s-r5" x="0" y="947.2" textLength="12.2" clip-path="url(#breeze-k8s-line-38)">│</text><text class="breeze-k8s-r4" x="24.4" y="947.2" textLength="231.8" clip-path="url(#breeze-k8s-line-38)">k9s                </text><text class="breeze-k8s-r1" x="280.6" y="947.2" textLength="1159" clip-path="url(#breeze-k8s-line-38)">Run k9s tool. You can pass any  [...]
-</text><text class="breeze-k8s-r5" x="0" y="971.6" textLength="1464" clip-path="url(#breeze-k8s-line-39)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-r1" x="1464" y="971.6" textLength="12.2" clip-path="url(#breeze-k8s-line-39)">
+</text><text class="breeze-k8s-r5" x="0" y="800.8" textLength="12.2" clip-path="url(#breeze-k8s-line-32)">│</text><text class="breeze-k8s-r4" x="24.4" y="800.8" textLength="231.8" clip-path="url(#breeze-k8s-line-32)">setup-lang-sdk-test</text><text class="breeze-k8s-r1" x="280.6" y="800.8" textLength="1159" clip-path="url(#breeze-k8s-line-32)">Provision the lang-SDK (Go + Java) coordinator system test on an already-deployed  [...]
+</text><text class="breeze-k8s-r5" x="0" y="825.2" textLength="12.2" clip-path="url(#breeze-k8s-line-33)">│</text><text class="breeze-k8s-r1" x="280.6" y="825.2" textLength="1159" clip-path="url(#breeze-k8s-line-33)">KubernetesExecutor cluster: build artifacts, build + load the Java worker image, deploy        </text><text class="breeze-k8s-r5" x="1451.8" y="825.2" textLength="12.2" clip-path=" [...]
+</text><text class="breeze-k8s-r5" x="0" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-line-34)">│</text><text class="breeze-k8s-r1" x="280.6" y="849.6" textLength="1159" clip-path="url(#breeze-k8s-line-34)">localstack S3, upload artifacts + stub Dag, create config, and upgrade the Helm release. With  </text><text class="breeze-k8s-r5" x="1451.8" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s [...]
+</text><text class="breeze-k8s-r5" x="0" y="874" textLength="12.2" clip-path="url(#breeze-k8s-line-35)">│</text><text class="breeze-k8s-r4" x="280.6" y="874" textLength="292.8" clip-path="url(#breeze-k8s-line-35)">--ts-sdk-native-dag-test</text><text class="breeze-k8s-r1" x="573.4" y="874" textLength="866.2" clip-path="url(#breeze-k8s-line-35)"> it also builds the TypeScript bundle and worker image. Run the test    [...]
+</text><text class="breeze-k8s-r5" x="0" y="898.4" textLength="12.2" clip-path="url(#breeze-k8s-line-36)">│</text><text class="breeze-k8s-r1" x="280.6" y="898.4" textLength="756.4" clip-path="url(#breeze-k8s-line-36)">afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests </text><text class="breeze-k8s-r4" x="1037" y="898.4" textLength="122" clip-path="url(#breeze-k8s-line-36)">--executor</text><text class="breeze-k8s-r1" x="1159" y="898.4" textLength [...]
+</text><text class="breeze-k8s-r5" x="0" y="922.8" textLength="12.2" clip-path="url(#breeze-k8s-line-37)">│</text><text class="breeze-k8s-r6" x="280.6" y="922.8" textLength="24.4" clip-path="url(#breeze-k8s-line-37)">-k</text><text class="breeze-k8s-r1" x="305" y="922.8" textLength="1134.6" clip-path="url(#breeze-k8s-line-37)"> test_lang_sdk_combined_dag_succeeds`.                     [...]
+</text><text class="breeze-k8s-r5" x="0" y="947.2" textLength="12.2" clip-path="url(#breeze-k8s-line-38)">│</text><text class="breeze-k8s-r4" x="24.4" y="947.2" textLength="231.8" clip-path="url(#breeze-k8s-line-38)">shell              </text><text class="breeze-k8s-r1" x="280.6" y="947.2" textLength="1159" clip-path="url(#breeze-k8s-line-38)">Run shell environment for the current KinD [...]
+</text><text class="breeze-k8s-r5" x="0" y="971.6" textLength="12.2" clip-path="url(#breeze-k8s-line-39)">│</text><text class="breeze-k8s-r4" x="24.4" y="971.6" textLength="231.8" clip-path="url(#breeze-k8s-line-39)">k9s                </text><text class="breeze-k8s-r1" x="280.6" y="971.6" textLength="1159" clip-path="url(#breeze-k8s-line-39)">Run k9s tool. You can pass any  [...]
+</text><text class="breeze-k8s-r5" x="0" y="996" textLength="1464" clip-path="url(#breeze-k8s-line-40)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-r1" x="1464" y="996" textLength="12.2" clip-path="url(#breeze-k8s-line-40)">
</text>
</g>
</g>
diff --git a/dev/breeze/doc/images/output_k8s.txt b/dev/breeze/doc/images/output_k8s.txt
index b983d60ab98..b33ca1c53a8 100644
--- a/dev/breeze/doc/images/output_k8s.txt
+++ b/dev/breeze/doc/images/output_k8s.txt
@@ -1 +1 @@
-e6bce3a1cc0d3660f16669fe5c4b1338
+7275bb89187fba314464a6566f7bf7fe
diff --git a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg
index a9e646da6c4..cfecd2b0602 100644
--- a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg
+++ b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.svg
@@ -1,4 +1,4 @@
-<svg class="rich-terminal" viewBox="0 0 1482 684.4" xmlns="http://www.w3.org/2000/svg">
+<svg class="rich-terminal" viewBox="0 0 1482 952.8" xmlns="http://www.w3.org/2000/svg">
<!-- Generated with Rich https://www.textualize.io -->
<style>
@@ -43,7 +43,7 @@
<defs>
<clipPath id="breeze-k8s-setup-lang-sdk-test-clip-terminal">
- <rect x="0" y="0" width="1463.0" height="633.4" />
+ <rect x="0" y="0" width="1463.0" height="901.8" />
</clipPath>
<clipPath id="breeze-k8s-setup-lang-sdk-test-line-0">
<rect x="0" y="1.5" width="1464" height="24.65"/>
@@ -120,9 +120,42 @@
<clipPath id="breeze-k8s-setup-lang-sdk-test-line-24">
<rect x="0" y="587.1" width="1464" height="24.65"/>
</clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-25">
+ <rect x="0" y="611.5" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-26">
+ <rect x="0" y="635.9" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-27">
+ <rect x="0" y="660.3" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-28">
+ <rect x="0" y="684.7" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-29">
+ <rect x="0" y="709.1" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-30">
+ <rect x="0" y="733.5" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-31">
+ <rect x="0" y="757.9" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-32">
+ <rect x="0" y="782.3" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-33">
+ <rect x="0" y="806.7" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-34">
+ <rect x="0" y="831.1" width="1464" height="24.65"/>
+ </clipPath>
+<clipPath id="breeze-k8s-setup-lang-sdk-test-line-35">
+ <rect x="0" y="855.5" width="1464" height="24.65"/>
+ </clipPath>
</defs>
- <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="682.4" rx="8"/><text class="breeze-k8s-setup-lang-sdk-test-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: k8s setup-lang-sdk-test</text>
+ <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1" x="1" y="1" width="1480" height="950.8" rx="8"/><text class="breeze-k8s-setup-lang-sdk-test-title" fill="#c5c8c6" text-anchor="middle" x="740" y="27">Command: k8s setup-lang-sdk-test</text>
<g transform="translate(26,22)">
<circle cx="0" cy="0" r="7" fill="#ff5f57"/>
<circle cx="22" cy="0" r="7" fill="#febc2e"/>
@@ -135,29 +168,40 @@
<text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="20" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-0)">
</text><text class="breeze-k8s-setup-lang-sdk-test-r2" x="12.2" y="44.4" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-1)">Usage:</text><text class="breeze-k8s-setup-lang-sdk-test-r3" x="97.6" y="44.4" textLength="366" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-1)">breeze k8s setup-lang-sdk-test</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="475.8" y="44.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-1)">[</te [...]
</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="68.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-2)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="93.2" textLength="1390.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">Provision the lang-SDK (Go + Java + TypeScript) coordinator system test on an already-deployed KubernetesExecutor </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="93.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="117.6" textLength="1439.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-4)">cluster: build artifacts, build + load the Java and TypeScript worker images, deploy localstack S3, upload artifacts +</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="117.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test- [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="142" textLength="1378.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">stub Dag, create config, and upgrade the Helm release. Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="142" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="166.4" textLength="207.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">breeze k8s tests </text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="219.6" y="166.4" textLength="122" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">--executor</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="166.4" textLength="280.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)"> [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="190.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-7)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="215.2" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="215.2" textLength="305" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8)"> K8S lang-SDK test flags </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="329.4" y="215.2" textLength="1110.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8 [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="239.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="239.6" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9)">--python            </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="292.8" y="239.6" textLength="24.4" clip-path="url(#breeze-k8s [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="264" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-10)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="341.6" y="264" textLength="317.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-10)">3.11 | 3.12 | 3.13 | 3.14)</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="264" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-lin [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="288.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-11)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="288.4" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-11)">--kubernetes-version</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="288.4" textLength="671" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-11)">Kubernetes version  [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="312.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-12)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="341.6" y="312.8" textLength="597.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-12)">v1.31.12 | v1.32.8 | v1.33.4 | v1.34.0 | v1.35.0)</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="312.8" textLength="12.2" clip-path=" [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="337.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-13)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="337.2" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-13)">--go-image          </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="337.2" textLength="976" clip-path="url(#breeze-k8s-setup-la [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="361.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="361.6" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">--java-image        </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="361.6" textLength="1098" clip-path="url(#breeze-k8s-setup-lang-sdk-te [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="386" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="386" textLength="1098" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">the prod image plus a headless JRE (Dockerfile.java) and loading it into the kind cluster.</text><text class="breeze-k8s-setup-la [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="410.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="341.6" y="410.4" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">(TEXT)</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="410.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">│</text><text class="breeze-k8s-setup- [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="434.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="434.8" textLength="244" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">--ts-image          </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="434.8" textLength="1098" clip-path="url(#breeze-k8s-setup-l [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="459.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="459.2" textLength="1098" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">building the prod image plus Node (Dockerfile.typescript) and loading it into the kind    </text><text class="breez [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="483.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="341.6" y="483.6" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">cluster. </text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="451.4" y="483.6" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">(TEXT)</text><text class="bree [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="508" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="508" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="532.4" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="532.4" textLength="195.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)"> Common options </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="219.6" y="532.4" textLength="1220" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">───────────── [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="556.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="556.8" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">--verbose</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="556.8" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">-v</text><text class="breeze-k8s-set [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="581.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="581.2" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">--dry-run</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="581.2" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">-D</text><text class="breeze-k8s-set [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="605.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="605.6" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">--help   </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="605.6" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">-h</text><text class= [...]
-</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="630" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="630" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="93.2" textLength="1415.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">Provision the lang-SDK (Go + Java) coordinator system test on an already-deployed KubernetesExecutor cluster: build </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="93.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-3)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="117.6" textLength="1427.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-4)">artifacts, build + load the Java worker image, deploy localstack S3, upload artifacts + stub Dag, create config, and </text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="117.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang- [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="142" textLength="378.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">upgrade the Helm release. With </text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="390.4" y="142" textLength="292.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-5)">--ts-sdk-native-dag-test</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="683.2" y="142" textLength="768.6" clip-path="url(#breeze-k8 [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="166.4" textLength="817.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests </text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="829.6" y="166.4" textLength="122" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-6)">--executor</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="951.6" y="166.4" textLeng [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="12.2" y="190.8" textLength="451.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-7)">test_lang_sdk_combined_dag_succeeds`.</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="190.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-7)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="215.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-8)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="239.6" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="239.6" textLength="305" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9)"> K8S lang-SDK test flags </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="329.4" y="239.6" textLength="1110.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-9 [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="264" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-10)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="264" textLength="634.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-10)">--python                                 [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="288.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-11)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="288.4" textLength="97.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-11)">images. </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="829.6" y="288.4" textLength="183" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-11)">[default: 3.10]</text><text cl [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="312.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-12)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="732" y="312.8" textLength="61" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-12)">3.14)</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="312.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-12)">│</text><text class="breeze-k8s-setup-lang- [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="337.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-13)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="337.2" textLength="634.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-13)">--kubernetes-version                             & [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="361.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="732" y="361.6" textLength="231.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">[default: v1.30.13]</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="976" y="361.6" textLength="427" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-14)">(>v1.30.13< |& [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="386" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="732" y="386" textLength="341.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">v1.33.4 | v1.34.0 | v1.35.0)</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="386" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-15)">│</ [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="410.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="410.4" textLength="634.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-16)">--go-image                                [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="434.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="434.8" textLength="329.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-17)">Defaults to the k8s image. </text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="1061.4" y="434.8" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-lin [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="459.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="459.2" textLength="634.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-18)">--java-image                               [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="483.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="483.6" textLength="707.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-19)">include a JRE. Defaults to building the prod image plus a </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="483.6" textLen [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="508" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="508" textLength="707.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-20)">headless JRE (Dockerfile.java) and loading it into the    </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="508" textLength="1 [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="532.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="532.4" textLength="170.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">kind cluster. </text><text class="breeze-k8s-setup-lang-sdk-test-r7" x="902.8" y="532.4" textLength="73.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-21)">(TEXT)</text><text cla [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="556.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="556.8" textLength="634.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-22)">--ts-image                                [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="581.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="581.2" textLength="707.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-23)">Must include Node. Defaults to building the prod image    </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="581.2" te [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="605.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="605.6" textLength="707.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-24)">plus Node (Dockerfile.typescript) and loading it into the </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="605.6" textLength="12.2" clip [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="630" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="630" textLength="292.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">kind cluster. Used with </text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="1024.8" y="630" textLength="292.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-25)">--ts-s [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="654.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-26)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="654.4" textLength="292.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-26)">--ts-sdk-native-dag-test</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="317.2" y="654.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-26)">/</text><text class=" [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="678.8" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-27)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="678.8" textLength="707.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-27)">TypeScript bundle, which needs a Node toolchain, and build</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="678.8" textLength="12.2" clip [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="703.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-28)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="703.2" textLength="707.6" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-28)">the TypeScript worker image. Off by default, so the Go +  </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="703.2" te [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="727.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-29)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="732" y="727.6" textLength="427" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-29)">Java test needs no Node toolchain. </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1159" y="727.6" textLength="122" clip-path="url(#breeze-k8s-setup-lang-sdk- [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="752" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-30)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="732" y="752" textLength="317.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-30)">no-ts-sdk-native-dag-test]</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="1451.8" y="752" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-30)">│</text><text class="bree [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="776.4" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-31)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="776.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-31)">
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="800.8" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-32)">╭─</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="24.4" y="800.8" textLength="195.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-32)"> Common options </text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="219.6" y="800.8" textLength="1220" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-32)">───────────── [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="825.2" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-33)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="825.2" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-33)">--verbose</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="825.2" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-33)">-v</text><text class="breeze-k8s-set [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="849.6" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-34)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="849.6" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-34)">--dry-run</text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="849.6" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-34)">-D</text><text class="breeze-k8s-set [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="874" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-35)">│</text><text class="breeze-k8s-setup-lang-sdk-test-r4" x="24.4" y="874" textLength="109.8" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-35)">--help   </text><text class="breeze-k8s-setup-lang-sdk-test-r5" x="158.6" y="874" textLength="24.4" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-35)">-h</text><text class="breez [...]
+</text><text class="breeze-k8s-setup-lang-sdk-test-r6" x="0" y="898.4" textLength="1464" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-36)">╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯</text><text class="breeze-k8s-setup-lang-sdk-test-r1" x="1464" y="898.4" textLength="12.2" clip-path="url(#breeze-k8s-setup-lang-sdk-test-line-36)">
</text>
</g>
</g>
diff --git a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt
index c9d173a905d..1450c6d09a0 100644
--- a/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt
+++ b/dev/breeze/doc/images/output_k8s_setup-lang-sdk-test.txt
@@ -1 +1 @@
-6a9a0568a2b313fb5a6780431efd2937
+2f5c7f33daf8d4b751977d96d8214fc3
diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
index 6b2eb800c5b..933c63b9d87 100644
--- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
+++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
@@ -2933,9 +2933,13 @@ def _lang_sdk_deploy_localstack(python: str, kubernetes_version: str, output: Ou
def _lang_sdk_upload_artifacts(
- staging: Path, python: str, kubernetes_version: str, output: Output | None
+ staging: Path, python: str, kubernetes_version: str, output: Output | None, *, typescript: bool = False
) -> None:
- """Copy artifacts + stub Dag into the localstack pod and create/fill S3 buckets via awslocal."""
+ """
+ Copy artifacts + stub Dag into the localstack pod and create/fill S3 buckets via awslocal.
+
+ ``typescript`` also uploads the TypeScript bundle and the Dag its native Dag triggers.
+ """
pod = run_command_with_k8s_env(
[
"kubectl",
@@ -2957,30 +2961,43 @@ def _lang_sdk_upload_artifacts(
).stdout.strip()
go_bundle = staging / "go-artifacts" / "lang_sdk_combined"
- ts_bundle = staging / "ts-artifacts" / LANG_SDK_TS_BUNDLE_NAME
if get_dry_run():
# The dry-run build steps produce no jar; use the placeholder name so the commands still print.
java_jar = staging / "java-artifacts" / "app.jar"
else:
java_jar = next((staging / "java-artifacts").glob("*.jar"))
stub_dag = LANG_SDK_PATH / "dags" / "lang_sdk_combined.py"
- # The native TypeScript Dag triggers this one, so it has to be in the
- # cluster for its trigger task to have a target. It is the same Python Dag
- # the mixed-language half of `ts-sdk/example` is written against.
- ts_stub_dag = LANG_SDK_TS_EXAMPLE_PATH / "dags" / "typescript_example.py"
-
- for src, dest in (
+ copies = [
(go_bundle, "/tmp/go_bundle"),
(java_jar, "/tmp/app.jar"),
- (ts_bundle, f"/tmp/{LANG_SDK_TS_BUNDLE_NAME}"),
(stub_dag, "/tmp/lang_sdk_combined.py"),
- (ts_stub_dag, "/tmp/typescript_example.py"),
- ):
+ ]
+ buckets = ["go-artifacts", "java-artifacts", "dags"]
+ uploads = [
+ ("/tmp/go_bundle", "s3://go-artifacts/lang_sdk_combined"),
+ ("/tmp/app.jar", "s3://java-artifacts/app.jar"),
+ ("/tmp/lang_sdk_combined.py", "s3://dags/lang_sdk_combined.py"),
+ ]
+ if typescript:
+ # The native TypeScript Dag triggers typescript_example, so it has to be in the
+ # cluster for its trigger task to have a target. It is the same Python Dag
+ # the mixed-language half of `ts-sdk/example` is written against.
+ copies += [
+ (staging / "ts-artifacts" / LANG_SDK_TS_BUNDLE_NAME, f"/tmp/{LANG_SDK_TS_BUNDLE_NAME}"),
+ (LANG_SDK_TS_EXAMPLE_PATH / "dags" / "typescript_example.py", "/tmp/typescript_example.py"),
+ ]
+ buckets.append("ts-artifacts")
+ uploads += [
+ (f"/tmp/{LANG_SDK_TS_BUNDLE_NAME}", f"s3://ts-artifacts/{LANG_SDK_TS_BUNDLE_NAME}"),
+ ("/tmp/typescript_example.py", "s3://dags/typescript_example.py"),
+ ]
+
+ for src, dest in copies:
_lang_sdk_kubectl(
["cp", str(src), f"{HELM_AIRFLOW_NAMESPACE}/{pod}:{dest}"], python, kubernetes_version, output
)
- for bucket in ("go-artifacts", "java-artifacts", "ts-artifacts", "dags"):
+ for bucket in buckets:
_lang_sdk_kubectl(
["exec", "-n", HELM_AIRFLOW_NAMESPACE, pod, "--", "awslocal", "s3", "mb", f"s3://{bucket}"],
python,
@@ -2988,13 +3005,6 @@ def _lang_sdk_upload_artifacts(
output,
check=False,
)
- uploads = (
- ("/tmp/go_bundle", "s3://go-artifacts/lang_sdk_combined"),
- ("/tmp/app.jar", "s3://java-artifacts/app.jar"),
- (f"/tmp/{LANG_SDK_TS_BUNDLE_NAME}", f"s3://ts-artifacts/{LANG_SDK_TS_BUNDLE_NAME}"),
- ("/tmp/lang_sdk_combined.py", "s3://dags/lang_sdk_combined.py"),
- ("/tmp/typescript_example.py", "s3://dags/typescript_example.py"),
- )
for src, dest in uploads:
_lang_sdk_kubectl(
["exec", "-n", HELM_AIRFLOW_NAMESPACE, pod, "--", "awslocal", "s3", "cp", src, dest],
@@ -3179,17 +3189,21 @@ def _setup_lang_sdk_test(
go_image: str | None = None,
java_image: str | None = None,
ts_image: str | None = None,
+ ts_sdk_native_dag_test: bool = False,
output: Output | None = None,
) -> None:
"""Provision the lang-SDK coordinator env on an already-deployed KubernetesExecutor cluster.
- Resolves the go-sdk/java-sdk sources, then builds the Go/Java/TypeScript artifacts, the Java
- and TypeScript worker images and deploys localstack in parallel, then serially uploads the
- artifacts, applies the config + secret, and helm-upgrades Airflow with the lang-SDK values.
+ Resolves the go-sdk/java-sdk sources, then builds the Go/Java artifacts, the Java worker
+ image and deploys localstack in parallel, then serially uploads the artifacts, applies the
+ config + secret, and helm-upgrades Airflow with the lang-SDK values.
+
+ ``ts_sdk_native_dag_test`` also builds the TypeScript bundle and worker image, which need a
+ Node toolchain, for the native TypeScript Dag test.
"""
go_image = go_image or f"{BuildProdParams(python=python).airflow_image_kubernetes}:latest"
build_java_image = java_image is None
- build_ts_image = ts_image is None
+ build_ts_image = ts_sdk_native_dag_test and ts_image is None
if java_image is None:
# The worker-image build below produces this fixed tag; resolve it up-front so the config
# rendering (which needs the tag, not the build result) does not depend on the parallel run.
@@ -3211,12 +3225,14 @@ def _setup_lang_sdk_test(
"Build Java jar",
lambda o: _lang_sdk_build_java_jar(staging, java_sdk_source, o, native=native),
),
- (
- "Build TypeScript bundle",
- lambda o: _lang_sdk_build_ts_bundle(staging, o, native=native),
- ),
- ("Deploy localstack", lambda o: _lang_sdk_deploy_localstack(python, kubernetes_version, o)),
]
+ if ts_sdk_native_dag_test:
+ steps.append(
+ ("Build TypeScript bundle", lambda o: _lang_sdk_build_ts_bundle(staging, o, native=native))
+ )
+ steps.append(
+ ("Deploy localstack", lambda o: _lang_sdk_deploy_localstack(python, kubernetes_version, o))
+ )
if build_java_image:
steps.append(
(
@@ -3232,17 +3248,19 @@ def _setup_lang_sdk_test(
)
)
_run_lang_sdk_parallel(steps, output=output)
- _lang_sdk_upload_artifacts(staging, python, kubernetes_version, output)
+ _lang_sdk_upload_artifacts(
+ staging, python, kubernetes_version, output, typescript=ts_sdk_native_dag_test
+ )
_lang_sdk_apply_configmaps_and_secret(python, kubernetes_version, go_image, java_image, ts_image, output)
_lang_sdk_deploy_airflow(python, kubernetes_version, output)
@kubernetes_group.command(
name="setup-lang-sdk-test",
- help="Provision the lang-SDK (Go + Java + TypeScript) coordinator system test on an "
- "already-deployed KubernetesExecutor cluster: build artifacts, build + load the Java and "
- "TypeScript worker images, deploy localstack S3, upload artifacts + stub Dag, create config, "
- "and upgrade the Helm release. "
+ help="Provision the lang-SDK (Go + Java) coordinator system test on an already-deployed "
+ "KubernetesExecutor cluster: build artifacts, build + load the Java worker image, deploy "
+ "localstack S3, upload artifacts + stub Dag, create config, and upgrade the Helm release. "
+ "With --ts-sdk-native-dag-test it also builds the TypeScript bundle and worker image. "
"Run the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests "
"--executor KubernetesExecutor -- -k test_lang_sdk_combined_dag_succeeds`.",
)
@@ -3260,7 +3278,17 @@ def _setup_lang_sdk_test(
@click.option(
"--ts-image",
help="Image for the TypeScript (NodeCoordinator) worker pod. Must include Node. Defaults to "
- "building the prod image plus Node (Dockerfile.typescript) and loading it into the kind cluster.",
+ "building the prod image plus Node (Dockerfile.typescript) and loading it into the kind cluster. "
+ "Used with --ts-sdk-native-dag-test.",
+)
+@click.option(
+ "--ts-sdk-native-dag-test/--no-ts-sdk-native-dag-test",
+ help="Also provision the native TypeScript Dag test: pack the TypeScript bundle, which needs a "
+ "Node toolchain, and build the TypeScript worker image. Off by default, so the Go + Java test "
+ "needs no Node toolchain.",
+ default=False,
+ show_default=True,
+ envvar="RUN_TS_SDK_NATIVE_DAG_K8S_TESTS",
)
@option_verbose
@option_dry_run
@@ -3270,6 +3298,7 @@ def setup_lang_sdk_test(
go_image: str | None,
java_image: str | None,
ts_image: str | None,
+ ts_sdk_native_dag_test: bool,
):
result = sync_virtualenv(force_venv_setup=False)
if result.returncode != 0:
@@ -3281,6 +3310,7 @@ def setup_lang_sdk_test(
go_image=go_image,
java_image=java_image,
ts_image=ts_image,
+ ts_sdk_native_dag_test=ts_sdk_native_dag_test,
output=None,
)
console_print(
diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py
index b72db432868..3f8a7a56558 100644
--- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py
+++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands_config.py
@@ -57,6 +57,7 @@ KUBERNETES_PARAMETERS: dict[str, list[dict[str, str | list[str]]]] = {
"--go-image",
"--java-image",
"--ts-image",
+ "--ts-sdk-native-dag-test",
],
}
],
diff --git a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
index a94976a44e7..7a8276c2c35 100644
--- a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
+++ b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
@@ -19,6 +19,7 @@ from __future__ import annotations
from unittest import mock
import pytest
+from click.testing import CliRunner
from airflow_breeze.commands import kubernetes_commands
from airflow_breeze.commands.kubernetes_commands import (
@@ -377,6 +378,113 @@ class TestSetupLangSdkTestNativeSelection:
}
+class TestSetupLangSdkTestTypeScriptGate:
+ """The TypeScript artifacts need a Node toolchain, so only the native TypeScript Dag test builds them."""
+
+ @pytest.fixture
+ def recorded(self, monkeypatch, tmp_path):
+ recorded: dict[str, object] = {}
+
+ def fake_parallel(steps, output):
+ recorded["steps"] = [title for title, _thunk in steps]
+
+ monkeypatch.setattr(kubernetes_commands, "_run_lang_sdk_parallel", fake_parallel)
+ monkeypatch.setattr(
+ kubernetes_commands,
+ "_lang_sdk_resolve_sdk_sources",
+ lambda staging, output: (tmp_path / "go_sdk", tmp_path / "java_sdk"),
+ )
+ monkeypatch.setattr(
+ kubernetes_commands,
+ "_lang_sdk_upload_artifacts",
+ lambda *args, typescript: recorded.update(typescript=typescript),
+ )
+ for name in ("_lang_sdk_apply_configmaps_and_secret", "_lang_sdk_deploy_airflow"):
+ monkeypatch.setattr(kubernetes_commands, name, lambda *a, **k: None)
+ monkeypatch.setattr(
+ kubernetes_commands,
+ "BuildProdParams",
+ lambda python: mock.Mock(airflow_image_kubernetes="img"),
+ )
+ return recorded
+
+ @pytest.mark.parametrize(
+ ("kwargs", "expected_steps", "expected_typescript"),
+ [
+ pytest.param(
+ {},
+ ["Build Go bundle", "Build Java jar", "Deploy localstack", "Build Java worker image"],
+ False,
+ id="default-go-and-java-only",
+ ),
+ pytest.param(
+ {"ts_sdk_native_dag_test": True},
+ [
+ "Build Go bundle",
+ "Build Java jar",
+ "Build TypeScript bundle",
+ "Deploy localstack",
+ "Build Java worker image",
+ "Build TypeScript worker image",
+ ],
+ True,
+ id="native-ts-dag-test",
+ ),
+ pytest.param(
+ {"ts_sdk_native_dag_test": True, "ts_image": "my-ts-worker:1"},
+ [
+ "Build Go bundle",
+ "Build Java jar",
+ "Build TypeScript bundle",
+ "Deploy localstack",
+ "Build Java worker image",
+ ],
+ True,
+ id="native-ts-dag-test-with-given-image",
+ ),
+ ],
+ )
+ def test_typescript_steps_follow_the_flag(self, recorded, kwargs, expected_steps, expected_typescript):
+ kubernetes_commands._setup_lang_sdk_test(python="3.10", kubernetes_version="v1.35.0", **kwargs)
+
+ assert recorded == {"steps": expected_steps, "typescript": expected_typescript}
+
+ @pytest.mark.parametrize("typescript", [False, True])
+ @mock.patch.object(kubernetes_commands, "run_command_with_k8s_env")
+ def test_upload_artifacts_uploads_typescript_only_when_asked(
+ self, mock_run, typescript, dry_run, tmp_path
+ ):
+ mock_run.return_value = mock.Mock(stdout="")
+
+ _lang_sdk_upload_artifacts(tmp_path, "3.10", "v1.35.0", None, typescript=typescript)
+
+ commands = [" ".join(map(str, call.args[0])) for call in mock_run.call_args_list]
+ assert any("ts-artifacts" in command for command in commands) is typescript
+ assert any("typescript_example.py" in command for command in commands) is typescript
+
+ @pytest.mark.parametrize(
+ ("args", "env", "expected"),
+ [
+ pytest.param([], {}, False, id="default"),
+ pytest.param(["--ts-sdk-native-dag-test"], {}, True, id="flag"),
+ pytest.param([], {"RUN_TS_SDK_NATIVE_DAG_K8S_TESTS": "true"}, True, id="env"),
+ ],
+ )
+ @mock.patch.object(kubernetes_commands, "make_sure_kubernetes_tools_are_installed")
+ @mock.patch.object(kubernetes_commands, "sync_virtualenv")
+ @mock.patch.object(kubernetes_commands, "_setup_lang_sdk_test")
+ def test_command_passes_the_flag(
+ self, mock_setup, mock_sync, _mock_tools, args, env, expected, monkeypatch
+ ):
+ monkeypatch.delenv("RUN_TS_SDK_NATIVE_DAG_K8S_TESTS", raising=False)
+ mock_sync.return_value = mock.Mock(returncode=0)
+
+ result = CliRunner().invoke(kubernetes_commands.setup_lang_sdk_test, args, env=env)
+
+ assert result.exit_code == 0, result.output
+ assert mock_setup.call_args.kwargs["ts_sdk_native_dag_test"] is expected
+
+
class TestLangSdkResolveSdkSources:
@pytest.fixture
def repo_root(self, tmp_path, monkeypatch):
(airflow) 03/08: Check the native TypeScript Dag's source and triggered run
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch jason/lang-sdk-e2e/08-native-e2e
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit bac79752c037ebebb20d8831a944256a1dba6fc3
Author: ZHE YOU LIU <zh...@gmail.com>
AuthorDate: Mon Sep 28 07:46:10 2026 +0000
Check the native TypeScript Dag's source and triggered run
The Code view must show the bundle's entry module, and the Python task of
the native Dag must trigger typescript_example with its rendered conf.
---
.../airflow_e2e_tests/e2e_test_utils/clients.py | 7 ++++++
.../ts_sdk_tests/test_ts_sdk_native_dag.py | 26 ++++++++++++++++++++++
2 files changed, 33 insertions(+)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
index 803edf515e7..ee541f97f7d 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/clients.py
@@ -179,6 +179,13 @@ class AirflowClient:
"""List a Dag's tasks, with the edges each one carries."""
return self._make_request(method="GET", endpoint=f"dags/{dag_id}/tasks")
+ def get_dag_source(self, dag_id: str):
+ """Get the source code stored for a Dag's latest version."""
+ return self._make_request(method="GET", endpoint=f"dagSources/{dag_id}")
+
+ def get_dag_run(self, dag_id: str, run_id: str):
+ return self._make_request(method="GET", endpoint=f"dags/{dag_id}/dagRuns/{run_id}")
+
def trigger_dag_and_wait(self, dag_id: str, json=None):
"""Trigger a DAG and wait for it to complete."""
self.un_pause_dag(dag_id)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
index f8aa1f48a72..f4d304e17e8 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py
@@ -46,6 +46,8 @@ from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
_TS_TASK_TIMEOUT = 600
_DAG_ID = "typescript_native_example"
+# The Dag trigger_downstream starts; see the triggerDagRun call in ts-sdk/example/src/native.ts.
+_DOWNSTREAM_DAG_ID = "typescript_example"
# Read by the handlers; see ts-sdk/example/src/native.ts.
_NORTH_ROWS_VARIABLE = "typescript_native_north_rows"
@@ -89,6 +91,8 @@ def completed_run(parsed_dag: AirflowClient) -> _CompletedRun:
):
client.set_variable(key, value)
+ # Dags are paused at creation here, so the run trigger_downstream starts would stay queued.
+ client.un_pause_dag(_DOWNSTREAM_DAG_ID)
client.un_pause_dag(_DAG_ID)
resp = client.trigger_dag(_DAG_ID, json={"logical_date": datetime.now(timezone.utc).isoformat()})
run_id = resp["dag_run_id"]
@@ -113,6 +117,14 @@ def test_the_dag_the_bundle_parsed_is_registered(parsed_dag: AirflowClient):
assert {tag["name"] for tag in dag.get("tags") or []} >= {"typescript", "native"}
+def test_the_dag_source_is_the_bundle_entry_module(parsed_dag: AirflowClient):
+ """The Code view shows the TypeScript the bundle was packed from, not the minified bundle."""
+ content = parsed_dag.get_dag_source(_DAG_ID)["content"]
+
+ assert "new Bundle()" in content
+ assert 'from "./native.js"' in content
+
+
def test_the_graph_carries_every_construct(parsed_dag: AirflowClient):
"""Group prefixes, the fan-in, the branches and the trigger all survive parsing."""
tasks = parsed_dag.get_tasks(_DAG_ID).get("tasks", [])
@@ -176,6 +188,20 @@ def test_every_other_task_succeeded(completed_run: _CompletedRun):
)
+def test_the_trigger_started_the_downstream_run(completed_run: _CompletedRun):
+ """A Python worker rebuilt ``trigger_downstream`` from the bundle and it triggered the Dag."""
+ run_id = completed_run.xcom("trigger_downstream", key="trigger_run_id")
+ client = completed_run.client
+
+ state = client.wait_for_dag_run(dag_id=_DOWNSTREAM_DAG_ID, run_id=run_id, timeout=_TS_TASK_TIMEOUT)
+ run = client.get_dag_run(_DOWNSTREAM_DAG_ID, run_id)
+
+ assert state == "success", f"expected the downstream run to succeed; got {run!r}"
+ assert run["run_type"] == "operator_triggered"
+ # The conf template is rendered on the worker, from the rebuilt operator.
+ assert run["conf"] == {"triggered_by": _DAG_ID}
+
+
def test_xcoms_flow_between_typescript_tasks(completed_run: _CompletedRun):
"""The fan-in's arguments reached the handler, and its own push came back."""
assert completed_run.xcom("extract.north") == {"region": "north", "rows": 3}