From e9efa4709192d03795ba513f6ff56bbf64787cf0 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Tue, 29 Sep 2026 12:11:55 +0000 Subject: [PATCH 01/28] TS SDK: cover native TypeScript Dags end to end --- .../e2e_test_utils/clients.py | 22 ++ .../ts_sdk_tests/test_ts_sdk_native_dag.py | 239 ++++++++++++++++++ dev/breeze/doc/images/output_k8s.svg | 8 +- dev/breeze/doc/images/output_k8s.txt | 2 +- .../images/output_k8s_setup-lang-sdk-test.svg | 38 ++- .../images/output_k8s_setup-lang-sdk-test.txt | 2 +- .../commands/kubernetes_commands.py | 165 +++++++++++- .../commands/kubernetes_commands_config.py | 1 + .../lang_sdk/Dockerfile.typescript | 41 +++ kubernetes-tests/lang_sdk/README.md | 18 +- kubernetes-tests/lang_sdk/config/values.yaml | 20 +- .../pod_templates/lang_sdk_typescript.yaml | 115 +++++++++ .../test_lang_sdk_coordinator_executor.py | 89 +++++++ ts-sdk/adr/0002-native-dag-interface.md | 18 +- ts-sdk/example/src/main.ts | 12 +- ts-sdk/example/src/native.ts | 168 ++++++++++++ ts-sdk/tests/sdk/native-dag-example.test.ts | 215 ++++++++++++++++ ts-sdk/tsconfig.json | 10 +- ts-sdk/vitest.config.ts | 11 + 19 files changed, 1141 insertions(+), 53 deletions(-) create mode 100644 airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py create mode 100644 kubernetes-tests/lang_sdk/Dockerfile.typescript create mode 100644 kubernetes-tests/lang_sdk/pod_templates/lang_sdk_typescript.yaml create mode 100644 ts-sdk/example/src/native.ts create mode 100644 ts-sdk/tests/sdk/native-dag-example.test.ts 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 7f85f07332210..c5d290a7c835d 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 @@ -192,6 +192,21 @@ 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 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) @@ -253,6 +268,13 @@ def get_task_logs( endpoint=endpoint, ) + def get_event_logs(self, dag_id: str, run_id: str, task_id: str | None = None) -> list[dict]: + """List the audit log events of a Dag run, or of one of its tasks, oldest first.""" + params = {"dag_id": dag_id, "run_id": run_id, "order_by": "event_log_id"} + if task_id is not None: + params["task_id"] = task_id + return self._make_request(method="GET", endpoint="eventLogs", params=params)["event_logs"] + class TaskSDKClient: """Client for interacting with the Task SDK API.""" 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 0000000000000..e5d639a26488b --- /dev/null +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_native_dag.py @@ -0,0 +1,239 @@ +# 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 task that triggers another +Dag's run. + +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" +# 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" +_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) + + # 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"] + 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 + } + # The trigger reads as the operator it mirrors, though the Node coordinator runs it. + 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"]} + + +def test_the_trigger_deferred_until_the_downstream_run_finished(completed_run: _CompletedRun): + """It deferred to ``DagStateTrigger``, which the Python triggerer runs, and resumed in TypeScript.""" + client = completed_run.client + events = client.get_event_logs(dag_id=_DAG_ID, run_id=completed_run.run_id, task_id="trigger_downstream") + lifecycle = [event for event in events if event["event"] in {"running", "deferred", "success"}] + + # Two starts of one try: the second is the resume, not a retry. + assert [(event["event"], event["try_number"]) for event in lifecycle] == [ + ("running", 1), + ("deferred", 1), + ("running", 1), + ("success", 1), + ], f"trigger_downstream events: {events}" + + # The queue the TypeScript coordinator serves, which the resumed task instance kept. + tis = client.get_task_instances(dag_id=_DAG_ID, run_id=completed_run.run_id)["task_instances"] + assert next(ti for ti in tis if ti["task_id"] == "trigger_downstream")["queue"] == "typescript" + + # Event ids grow in the order the events are written, so the resume came after every event of + # the run it waited on. + downstream_run_id = completed_run.xcom("trigger_downstream", key="trigger_run_id") + downstream = client.get_event_logs(dag_id=_DOWNSTREAM_DAG_ID, run_id=downstream_run_id) + assert downstream, f"no events for {_DOWNSTREAM_DAG_ID} run {downstream_run_id}" + assert max(event["event_log_id"] for event in downstream) < lifecycle[2]["event_log_id"] diff --git a/dev/breeze/doc/images/output_k8s.svg b/dev/breeze/doc/images/output_k8s.svg index 085b233e8aa3e..92f835a53e8e3 100644 --- a/dev/breeze/doc/images/output_k8s.svg +++ b/dev/breeze/doc/images/output_k8s.svg @@ -206,10 +206,10 @@ │resources, and run the optional per-overlay pytest module.                                     │ │run-complete-tests Run complete k8s tests consisting of: creating cluster, building and uploading image, deploying│ │airflow, running tests and deleting clusters (optionally for all clusters in parallel).        │ -│setup-lang-sdk-testProvision 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. Run   │ -│the test afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests --executor│ +│setup-lang-sdk-testProvision 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`.                                 │ │shell              Run shell environment for the current KinD cluster.                                            │ │k9s                Run k9s tool. You can pass any k9s args as extra args.                                         │ diff --git a/dev/breeze/doc/images/output_k8s.txt b/dev/breeze/doc/images/output_k8s.txt index d0c3808909336..b983d60ab98ed 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 cec71c785fef4..e04accc756cfc 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 @@ - +

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 0000000000000..e00dccb27b5c5 --- /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 c060b88e00be1..23e2f67fd6e7f 100644 --- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py +++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py @@ -50,6 +50,8 @@ GO_SDK_STATE_STORE_RETENTION_DAYS, 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, @@ -402,8 +404,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 @@ -418,14 +420,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 = [ @@ -443,6 +448,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") @@ -456,7 +463,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 @@ -509,10 +516,16 @@ def _setup_java_sdk_integration(dot_env_file, tmp_dir): "classpath": "airflow.sdk.coordinators.java.JavaCoordinator", "kwargs": {"task_handler_bundle_name": "java-test-task-handlers"}, }, + # Serves the Dags folder: it parses the native-Dag JAR there and runs + # that JAR's tasks. + "java-native": { + "classpath": "airflow.sdk.coordinators.java.JavaCoordinator", + "kwargs": {"task_handler_bundle_name": "dags-folder"}, + }, } ) 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 87e7fa5bdf6f0..024a90fb01bec 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_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 0000000000000..f8c00b31fb3d1 --- /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}" From eced559d6be5af542752e4b2b16c2baf228955a5 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 28 Sep 2026 08:02:53 +0000 Subject: [PATCH 06/28] 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 +++++-- .../airflow_breeze/utils/selective_checks.py | 13 +++++ dev/breeze/tests/test_selective_checks.py | 47 +++++++++++++++++++ 3 files changed, 71 insertions(+), 5 deletions(-) diff --git a/dev/breeze/doc/ci/04_selective_checks.md b/dev/breeze/doc/ci/04_selective_checks.md index f75ab8d37cfa4..b383a9b850839 100644 --- a/dev/breeze/doc/ci/04_selective_checks.md +++ b/dev/breeze/doc/ci/04_selective_checks.md @@ -470,11 +470,17 @@ together using `pytest-xdist` (pytest-xdist distributes the tests among parallel canary and the changed tests must pass after it * `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`, `lang_sdk_processor.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`; + 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 @@ -640,7 +646,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 7a7d90b0fb1c0..0f0e46c9e21c2 100644 --- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py +++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py @@ -197,6 +197,16 @@ def __hash__(self): return hash(frozenset(self)) +# Core and Task SDK sources on the native Lang-SDK Dag path: discovery, parsing through a +# coordinator's runtime, importing it into a Dag bag, and reading its source. +LANG_SDK_NATIVE_DAG_PARSING_FILES = ( + r"^airflow-core/src/airflow/dag_processing/(dagbag|importer_routing|lang_sdk_processor|manager|processor)\.py$", + r"^airflow-core/src/airflow/models/dagcode\.py$", + r"^task-sdk/src/airflow/sdk/coordinators/_dag_importer\.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: [ @@ -263,11 +273,13 @@ def __hash__(self): # `.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. @@ -300,6 +312,7 @@ def __hash__(self): 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 28664c92413a3..3f15ba9e3d4a2 100644 --- a/dev/breeze/tests/test_selective_checks.py +++ b/dev/breeze/tests/test_selective_checks.py @@ -1612,6 +1612,53 @@ 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-core/src/airflow/dag_processing/lang_sdk_processor.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 Lang-SDK Dag file processor 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",), { From 0091f0032b51558854067e81dd63402d2a26b0d2 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 28 Sep 2026 09:02:43 +0000 Subject: [PATCH 07/28] 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. --- .../org/apache/airflow/e2e/NativeBundleBuilder.java | 2 +- .../apache/airflow/e2e/nativedag/AnnotationDag.java | 12 +++++++----- .../airflow_e2e_tests/e2e_test_utils/clients.py | 1 + .../java_sdk_tests/test_java_sdk_native_dag.py | 7 ++++++- 4 files changed, 15 insertions(+), 7 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 17ac91e4cd357..cfddd691de270 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 @@ * {@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 e00dccb27b5c5..89abab7737b2e 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 69fc7e42c1072..437917698c133 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 @@ -218,6 +218,7 @@ def get_dag_source(self, dag_id: str): 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 f8c00b31fb3d1..9dbc9ab87bc4a 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. + + The Code view shows the JAR's entrypoint, 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 From 3c61f55f337583430562f063ab62786b934423ea Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 28 Sep 2026 12:34:28 +0000 Subject: [PATCH 08/28] 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 +- .../images/output_k8s_setup-lang-sdk-test.svg | 92 ++++++++++----- .../images/output_k8s_setup-lang-sdk-test.txt | 2 +- .../commands/kubernetes_commands.py | 100 ++++++++++------ .../commands/kubernetes_commands_config.py | 1 + .../test_kubernetes_lang_sdk_commands.py | 108 ++++++++++++++++++ 7 files changed, 257 insertions(+), 74 deletions(-) diff --git a/dev/breeze/doc/images/output_k8s.svg b/dev/breeze/doc/images/output_k8s.svg index 92f835a53e8e3..8c3c725e05ca9 100644 --- a/dev/breeze/doc/images/output_k8s.svg +++ b/dev/breeze/doc/images/output_k8s.svg @@ -1,4 +1,4 @@ - +

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. + * {@code queue}, which {@code queue_to_coordinator} routes to the {@code java-jdk} coordinator. + * The {@code java-native} coordinator only parses this JAR. */ public class NativeBundleBuilder { public static final String QUEUE = "java-native"; From 9be77c222ec0f61dff91e5b8dff6c86a466d2f85 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Sat, 3 Oct 2026 18:05:51 +0000 Subject: [PATCH 18/28] Read a native Dag example's served Dags with bundleDags The test called listBundleNativeDags, which main's TypeScript SDK does not have. bundleDags returns the same Dags, keyed by dag_id. --- ts-sdk/tests/sdk/native-dag-example.test.ts | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/ts-sdk/tests/sdk/native-dag-example.test.ts b/ts-sdk/tests/sdk/native-dag-example.test.ts index e876d5ef29bd1..2c01faeb9addb 100644 --- a/ts-sdk/tests/sdk/native-dag-example.test.ts +++ b/ts-sdk/tests/sdk/native-dag-example.test.ts @@ -26,7 +26,7 @@ import { describe, expect, it } from "vitest"; import { serializeDag } from "../../src/coordinator/serde.js"; -import { Bundle, listBundleNativeDags } from "../../src/sdk/bundle.js"; +import { Bundle, bundleDags } from "../../src/sdk/bundle.js"; import { finalizeDag, getDagOrderEdges, getDagTaskRecords, type Dag } from "../../src/sdk/dag.js"; import { dag } from "../../example/src/native.js"; @@ -77,9 +77,7 @@ describe("the native Dag example", () => { // registration is rebuilt here rather than imported. const bundle = new Bundle(dag); - expect(listBundleNativeDags(bundle).map((served) => served.dagId)).toEqual([ - "typescript_native_example", - ]); + expect([...bundleDags(bundle).keys()]).toEqual(["typescript_native_example"]); }); describe("serializes", () => { From 383ccfeeb70e5723d0141066c66f64731c57b999 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Mon, 5 Oct 2026 04:42:00 +0000 Subject: [PATCH 19/28] Keep the k8s lang-SDK values comment within the YAML line limit Adding ts-task-handlers to the comment that lists the task handler Dag bundles made its first line 114 characters long, over the 110 that yamllint allows. --- kubernetes-tests/lang_sdk/config/values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kubernetes-tests/lang_sdk/config/values.yaml b/kubernetes-tests/lang_sdk/config/values.yaml index f850fdd21124d..5c35c5aae34e8 100644 --- a/kubernetes-tests/lang_sdk/config/values.yaml +++ b/kubernetes-tests/lang_sdk/config/values.yaml @@ -51,7 +51,7 @@ scheduler: # config.dag_processor.dag_bundle_config_list directly -- the chart renders the config from this and # it is the recommended, drift-safe path. # -# go-task-handlers / java-task-handlers / ts-task-handlers are the coordinators' task_handler_bundle_name bundles: +# go-task-handlers / java-task-handlers / ts-task-handlers are the task_handler_bundle_name bundles: # the worker pod's init container stages each artifact bucket into these paths (stage_artifacts.py). # They exist only in the worker pods, so the dag-processor finds no files in them. dagProcessor: From 18a9601e2632ad500ec31380e5d5463a65f1ed1a Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 5 Oct 2026 10:40:52 +0000 Subject: [PATCH 20/28] Rework the lang-SDK k8s harness for mixed-language and native Dag tests Renames the combined Go + Java system test to mixed-language and adds TypeScript to it, adds one native Dag test per language SDK (Java and TypeScript; Go is skipped, native Dag parsing has no Go support yet), and drops the TypeScript-only gate: every lang-SDK artifact now builds unconditionally. Every coordinator queue now shares one runtime image (prod + JRE + Node) and one worker pod template. A coordinator downloads its own task_handler_bundle_name bundle directly from S3, the same way any Python Dag bundle works, so the init container and stage_artifacts.py staging step are gone -- restoring the Go binary's execute bit after an S3 download is now ExecutableCoordinator's own job (see the executable-bundle-exec-bit PR this depends on). Every task-handler bucket carries an .airflowignore so the Dag processor never tries to parse a packed artifact as a Dag; the two native buckets carry none, since parsing them is the point. With one coordinator per language, no [sdk] dag_bundle_to_coordinator entry is needed. Also updates the k8s CI job to provision Node, and the breeze docs and unit tests to match. Co-Authored-By: Claude --- .github/workflows/k8s-tests.yml | 36 +- dev/breeze/doc/05_test_commands.rst | 10 +- dev/breeze/doc/images/output_k8s.svg | 12 +- dev/breeze/doc/images/output_k8s.txt | 2 +- .../images/output_k8s_run-complete-tests.svg | 6 +- .../images/output_k8s_run-complete-tests.txt | 2 +- .../images/output_k8s_setup-lang-sdk-test.svg | 94 +-- .../images/output_k8s_setup-lang-sdk-test.txt | 2 +- .../commands/kubernetes_commands.py | 533 +++++++++--------- .../commands/kubernetes_commands_config.py | 5 +- .../test_kubernetes_lang_sdk_commands.py | 209 +++---- kubernetes-tests/lang_sdk/Dockerfile.java | 31 - ...kerfile.typescript => Dockerfile.runtimes} | 16 +- kubernetes-tests/lang_sdk/README.md | 153 ++--- kubernetes-tests/lang_sdk/airflowignore_all | 1 + kubernetes-tests/lang_sdk/config/values.yaml | 67 ++- ...combined.py => lang_sdk_mixed_language.py} | 30 +- kubernetes-tests/lang_sdk/go_example/main.go | 21 +- .../airflow/k8sexample/K8sBundleBuilder.java | 4 +- ...Example.java => MixedLanguageExample.java} | 16 +- .../pod_templates/lang_sdk_golang.yaml | 118 ---- .../lang_sdk/pod_templates/lang_sdk_java.yaml | 116 ---- ...k_typescript.yaml => lang_sdk_worker.yaml} | 59 +- kubernetes-tests/lang_sdk/stage_artifacts.py | 92 --- .../lang_sdk/ts_example/.gitignore | 3 + .../lang_sdk/ts_example/package.json | 20 + .../lang_sdk/ts_example/pnpm-lock.yaml | 336 +++++++++++ .../lang_sdk/ts_example/src/main.ts | 47 ++ .../lang_sdk/ts_example/tsconfig.json | 25 + .../test_lang_sdk_mixed_language.py | 90 +++ ...xecutor.py => test_lang_sdk_native_dag.py} | 129 ++--- 31 files changed, 1208 insertions(+), 1077 deletions(-) delete mode 100644 kubernetes-tests/lang_sdk/Dockerfile.java rename kubernetes-tests/lang_sdk/{Dockerfile.typescript => Dockerfile.runtimes} (71%) create mode 100644 kubernetes-tests/lang_sdk/airflowignore_all rename kubernetes-tests/lang_sdk/dags/{lang_sdk_combined.py => lang_sdk_mixed_language.py} (68%) rename kubernetes-tests/lang_sdk/java_example/src/java/org/apache/airflow/k8sexample/{CombinedExample.java => MixedLanguageExample.java} (71%) delete mode 100644 kubernetes-tests/lang_sdk/pod_templates/lang_sdk_golang.yaml delete mode 100644 kubernetes-tests/lang_sdk/pod_templates/lang_sdk_java.yaml rename kubernetes-tests/lang_sdk/pod_templates/{lang_sdk_typescript.yaml => lang_sdk_worker.yaml} (58%) delete mode 100644 kubernetes-tests/lang_sdk/stage_artifacts.py create mode 100644 kubernetes-tests/lang_sdk/ts_example/.gitignore create mode 100644 kubernetes-tests/lang_sdk/ts_example/package.json create mode 100644 kubernetes-tests/lang_sdk/ts_example/pnpm-lock.yaml create mode 100644 kubernetes-tests/lang_sdk/ts_example/src/main.ts create mode 100644 kubernetes-tests/lang_sdk/ts_example/tsconfig.json create mode 100644 kubernetes-tests/tests/kubernetes_tests/test_lang_sdk_mixed_language.py rename kubernetes-tests/tests/kubernetes_tests/{test_lang_sdk_coordinator_executor.py => test_lang_sdk_native_dag.py} (60%) diff --git a/.github/workflows/k8s-tests.yml b/.github/workflows/k8s-tests.yml index ecad4e15e02ec..e4e3a6147ae7f 100644 --- a/.github/workflows/k8s-tests.yml +++ b/.github/workflows/k8s-tests.yml @@ -106,7 +106,7 @@ jobs: # preparing k8s environment with uv takes < 15 seconds with `uv` - there is no point in caching it. # The lang-SDK coordinator system test is intentionally NOT run in this matrix. It has its own # dedicated single-combo job (tests-kubernetes-lang-sdk below), so it no longer adds the ~6 minutes - # of Go/Java build + coordinator run to each of these system-test jobs. + # of Go/Java/TypeScript build + coordinator run to each of these system-test jobs. - name: "\ Run complete K8S tests ${{ matrix.executor }}-${{ env.PYTHON_MAJOR_MINOR_VERSION }}-\ ${{env.KUBERNETES_VERSION}}-${{ matrix.use-standard-naming }}" @@ -140,11 +140,12 @@ jobs: run: breeze k8s delete-cluster --all if: always() - # The Multi-Lang KubernetesExecutor coordinator system test builds a Go bundle and a Java jar and runs - # them as workers under KubernetesExecutor. It is executor/naming/version-independent, so instead of - # bolting it onto all six ``KubernetesExecutor``/``use-standard-naming == false`` system-test jobs above - # (which ran it redundantly and added ~6 minutes each), it runs once here on a single default combo and - # executes only the lang-SDK test (``-k test_lang_sdk_combined_dag_succeeds``) - not the full suite. + # The lang-SDK KubernetesExecutor coordinator system tests build a Go bundle, Java jars and + # TypeScript bundles and run them as workers under KubernetesExecutor. They are + # executor/naming/version-independent, so instead of bolting them onto all six + # ``KubernetesExecutor``/``use-standard-naming == false`` system-test jobs above (which ran them + # redundantly and added several minutes each), they run once here on a single default combo and + # execute only the lang-SDK tests (``-k TestLangSdk``) - not the full suite. tests-kubernetes-lang-sdk: timeout-minutes: 60 name: "K8S Lang-SDK:${{ inputs.lang-sdk-kubernetes-combo }}" @@ -182,7 +183,7 @@ jobs: use-uv: ${{ inputs.use-uv }} make-mnt-writeable-and-cleanup: true id: breeze - # Provision the lang-SDK Go/Java toolchains on the host so breeze builds the artifacts natively + # Provision the lang-SDK Go/Java/Node toolchains on the host so breeze builds the artifacts natively # (LANG_SDK_NATIVE_TOOLCHAIN=true) instead of in throwaway toolchain containers, skipping the image # pulls and reusing the caches restored below. The module/build and Gradle caches are keyed # explicitly (rather than via the setup-* built-in caching) so the key carries an ``-vN-`` salt: @@ -213,15 +214,28 @@ jobs: path: | ~/.gradle/caches ~/.gradle/wrapper - key: lang-sdk-gradle-v1-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('java-sdk/**/*.gradle*', 'java-sdk/**/gradle-wrapper.properties', 'kubernetes-tests/lang_sdk/java_example/**/*.gradle*') }} # yamllint disable-line rule:line-length + key: lang-sdk-gradle-v1-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('java-sdk/**/*.gradle*', 'java-sdk/**/gradle-wrapper.properties', 'kubernetes-tests/lang_sdk/java_example/**/*.gradle*', 'airflow-e2e-tests/java-native-bundle/**/*.gradle*') }} # yamllint disable-line rule:line-length restore-keys: | lang-sdk-gradle-v1-${{ runner.os }}-${{ runner.arch }}- + - name: "Setup pnpm for the lang-SDK TypeScript build" + uses: pnpm/action-setup@ea17c68df8912ef543352723c149a84f56e3d413 # v6.1.0 + with: + package_json_file: ts-sdk/package.json + run_install: false + - name: "Setup Node for the lang-SDK TypeScript build" + uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0 + with: + node-version: 22 + cache: 'pnpm' + cache-dependency-path: | + ts-sdk/pnpm-lock.yaml + kubernetes-tests/lang_sdk/ts_example/pnpm-lock.yaml # No --upgrade here: the redeploy/upgrade path is already exercised across the main matrix above; - # this job only needs a single deploy to validate the Multi-Lang coordinator, saving ~1.5 minutes. - - name: "Run lang-SDK K8S test ${{ inputs.lang-sdk-kubernetes-combo }}" + # this job only needs a single deploy to validate the lang-SDK coordinators, saving ~1.5 minutes. + - name: "Run lang-SDK K8S tests ${{ inputs.lang-sdk-kubernetes-combo }}" run: >- breeze k8s run-complete-tests --no-copy-local-sources - -- -k test_lang_sdk_combined_dag_succeeds + -- -k TestLangSdk env: EXECUTOR: "KubernetesExecutor" USE_STANDARD_NAMING: "false" diff --git a/dev/breeze/doc/05_test_commands.rst b/dev/breeze/doc/05_test_commands.rst index 111344c11ea80..980caf471f05b 100644 --- a/dev/breeze/doc/05_test_commands.rst +++ b/dev/breeze/doc/05_test_commands.rst @@ -591,11 +591,11 @@ Setting up the lang-SDK coordinator system test ............................................... ``breeze k8s setup-lang-sdk-test`` provisions a cluster for the lang-SDK coordinator -system test: it builds the Go and Java example bundles, deploys an in-cluster localstack -S3, uploads the artifacts and the Python stub Dag to their buckets, renders the -coordinator pod-template image placeholders, and installs the Helm release configured for -the ``golang`` and ``java`` queues. After it completes, run the test with -``breeze k8s tests``. +system tests: it builds the Go, Java and TypeScript example bundles (mixed-language and +native), deploys an in-cluster localstack S3, uploads the artifacts and the Dag files to +their buckets, builds and loads the shared runtime image (prod + JRE + Node), and installs +the Helm release configured for the ``golang``, ``java``, ``java-native`` and ``typescript`` +queues. After it completes, run the tests with ``breeze k8s tests``. All parameters of the command are here: diff --git a/dev/breeze/doc/images/output_k8s.svg b/dev/breeze/doc/images/output_k8s.svg index 8c3c725e05ca9..6ba5d750acf29 100644 --- a/dev/breeze/doc/images/output_k8s.svg +++ b/dev/breeze/doc/images/output_k8s.svg @@ -209,12 +209,12 @@ │resources, and run the optional per-overlay pytest module.                                     │ │run-complete-tests Run complete k8s tests consisting of: creating cluster, building and uploading image, deploying│ │airflow, running tests and deleting clusters (optionally for all clusters in parallel).        │ -│setup-lang-sdk-testProvision 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`.                                                       │ +│setup-lang-sdk-testProvision the lang-SDK (Go + Java + TypeScript) coordinator system tests on an already-deployed│ +│KubernetesExecutor cluster: build the Go bundle, the Java and TypeScript jars/bundles          │ +│(mixed-language and native), build + load the shared runtime image, deploy localstack S3,      │ +│upload artifacts + Dag files, create the pod template + secret, and upgrade the Helm release.  │ +│Run the tests afterwards with `RUN_LANG_SDK_K8S_TESTS=true breeze k8s tests --executor│ +│KubernetesExecutor -- -k TestLangSdk`.                                                         │ │shell              Run shell environment for the current KinD cluster.                                            │ │k9s                Run k9s tool. You can pass any k9s args as extra args.                                         │ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ diff --git a/dev/breeze/doc/images/output_k8s.txt b/dev/breeze/doc/images/output_k8s.txt index b33ca1c53a800..f48d1250bcdfb 100644 --- a/dev/breeze/doc/images/output_k8s.txt +++ b/dev/breeze/doc/images/output_k8s.txt @@ -1 +1 @@ -7275bb89187fba314464a6566f7bf7fe +e9520ded47d3061a2ee5357531b97d3e diff --git a/dev/breeze/doc/images/output_k8s_run-complete-tests.svg b/dev/breeze/doc/images/output_k8s_run-complete-tests.svg index 20f226260c9f5..61a6ae242c121 100644 --- a/dev/breeze/doc/images/output_k8s_run-complete-tests.svg +++ b/dev/breeze/doc/images/output_k8s_run-complete-tests.svg @@ -232,9 +232,9 @@ │--executor                        Executor to use for a kubernetes cluster. [default: LocalExecutor]│ │(>LocalExecutor< | KubernetesExecutor | CeleryExecutor | │ │CeleryKubernetesExecutor | EdgeExecutor)│ -│--lang-sdk-test/--no-lang-sdk-testProvision the lang-SDK (Go + Java) coordinator env and run its system test  │ -│as part of the run (KubernetesExecutor only). Off by default so the regular │ -│k8s suites skip the test. [default: no-lang-sdk-test]│ +│--lang-sdk-test/--no-lang-sdk-testProvision the lang-SDK (Go + Java + TypeScript) coordinator env and run its │ +│system tests as part of the run (KubernetesExecutor only). Off by default so│ +│the regular k8s suites skip the test. [default: no-lang-sdk-test]│ │--force-venv-setup                Force recreation of the virtualenv.│ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─ Parallel options ───────────────────────────────────────────────────────────────────────────────────────────────────╮ diff --git a/dev/breeze/doc/images/output_k8s_run-complete-tests.txt b/dev/breeze/doc/images/output_k8s_run-complete-tests.txt index fb169e12ae2b4..c2123bc7d6b8f 100644 --- a/dev/breeze/doc/images/output_k8s_run-complete-tests.txt +++ b/dev/breeze/doc/images/output_k8s_run-complete-tests.txt @@ -1 +1 @@ -01e6652ed4a23dbc298becc7f7ac0fff +6dcbadc22bd67eeacc3e8be4fdf8f9bb 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 63e008a0197fd..e5325a0b2430c 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 @@ - +