This is an automated email from the ASF dual-hosted git repository.
jason810496 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new eb52b8b3511 [v3-4-test] Cover YAML-generated Java Dags in the java_sdk
e2e test (#74710) (#75002)
eb52b8b3511 is described below
commit eb52b8b35115f004960bd7810685e52b13f10835
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Oct 18 14:20:44 2026 +0100
[v3-4-test] Cover YAML-generated Java Dags in the java_sdk e2e test
(#74710) (#75002)
---
airflow-e2e-tests/docker/java.yml | 6 ++
airflow-e2e-tests/java-native-bundle/build.gradle | 1 +
.../apache/airflow/e2e/NativeBundleBuilder.java | 12 ++-
.../apache/airflow/e2e/nativedag/DynamicDags.java | 74 ++++++++++++++++++
.../e2e_test_utils/dynamic_dags.py | 71 +++++++++++++++++
.../go_sdk_tests/test_go_sdk_dynamic_dag.py | 47 +++---------
.../java_sdk_tests/test_java_sdk_dynamic_dag.py | 89 ++++++++++++++++++++++
7 files changed, 260 insertions(+), 40 deletions(-)
diff --git a/airflow-e2e-tests/docker/java.yml
b/airflow-e2e-tests/docker/java.yml
index 3204e395132..1dadf0e5f65 100644
--- a/airflow-e2e-tests/docker/java.yml
+++ b/airflow-e2e-tests/docker/java.yml
@@ -29,6 +29,12 @@
services:
airflow-dag-processor:
image: airflow-java-worker
+ environment:
+ # The dynamic Dag test swaps a YAML file and waits for the Dag IDs to
change. Re-parse the
+ # bundle every few seconds, and mark a Dag stale soon after it leaves
the YAML.
+ AIRFLOW__DAG_PROCESSOR__MIN_FILE_PROCESS_INTERVAL: '5'
+ AIRFLOW__DAG_PROCESSOR__STALE_DAG_THRESHOLD: '10'
+ AIRFLOW__SCHEDULER__PARSING_CLEANUP_INTERVAL: '5'
airflow-worker:
image: airflow-java-worker
diff --git a/airflow-e2e-tests/java-native-bundle/build.gradle
b/airflow-e2e-tests/java-native-bundle/build.gradle
index 0e572c073fc..59ff66f77af 100644
--- a/airflow-e2e-tests/java-native-bundle/build.gradle
+++ b/airflow-e2e-tests/java-native-bundle/build.gradle
@@ -30,6 +30,7 @@ dependencies {
annotationProcessor("org.apache.airflow:airflow-sdk-processor:${projectVersion}")
implementation("org.apache.airflow:airflow-sdk:${projectVersion}")
implementation("org.apache.airflow:airflow-sdk-jpl:${projectVersion}")
+
implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.22.2")
}
java {
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 b204371c372..0c52b3fe426 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
@@ -19,8 +19,11 @@
package org.apache.airflow.e2e;
+import java.io.IOException;
+import java.net.URISyntaxException;
import java.util.List;
import org.apache.airflow.e2e.nativedag.AnnotationDag;
+import org.apache.airflow.e2e.nativedag.DynamicDags;
import org.apache.airflow.e2e.nativedag.TargetDag;
import org.apache.airflow.sdk.*;
@@ -140,11 +143,14 @@ public class NativeBundleBuilder {
return dag;
}
- public static Bundle build() {
- return new
Bundle().register(buildDag()).register(AnnotationDag.class).register(TargetDag.build());
+ public static Bundle build() throws IOException, URISyntaxException {
+ var bundle =
+ new
Bundle().register(buildDag()).register(AnnotationDag.class).register(TargetDag.build());
+ DynamicDags.load().forEach(bundle::register);
+ return bundle;
}
- public static void main(String[] args) {
+ public static void main(String[] args) throws IOException,
URISyntaxException {
Server.create(args).serve(build());
}
}
diff --git
a/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/DynamicDags.java
b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/DynamicDags.java
new file mode 100644
index 00000000000..88f4cf7268f
--- /dev/null
+++
b/airflow-e2e-tests/java-native-bundle/src/java/org/apache/airflow/e2e/nativedag/DynamicDags.java
@@ -0,0 +1,74 @@
+/*
+ * 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 static org.apache.airflow.e2e.NativeBundleBuilder.QUEUE;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.airflow.sdk.*;
+
+/**
+ * Native Java Dags generated from the {@code dags.yaml} next to the bundle
JAR, one per entry.
+ *
+ * <p>Both the Dag processor and the worker run the JAR, and each run reads
the YAML again, so
+ * swapping only the YAML changes the Dags with no rebuild. Without a {@code
dags.yaml} next to the
+ * JAR, as when the Gradle plugin runs the main class at build time or before
the e2e test writes
+ * one, there are no Dags, and the bundle's other Dags still parse.
+ */
+public class DynamicDags {
+ private static final Map<String, String> GREETINGS = new HashMap<>();
+
+ public static class Greet implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ client.setXCom(GREETINGS.get(context.ti.dagId));
+ }
+ }
+
+ public static List<DagDef> load() throws IOException, URISyntaxException {
+ var jar =
Path.of(DynamicDags.class.getProtectionDomain().getCodeSource().getLocation().toURI());
+ var path = jar.resolveSibling("dags.yaml");
+ var dags = new ArrayList<DagDef>();
+ if (!Files.isRegularFile(path)) {
+ return dags;
+ }
+ for (var entry : new ObjectMapper(new
YAMLFactory()).readTree(path.toFile()).path("dags")) {
+ var dagId = entry.path("dag_id").asText();
+ GREETINGS.put(dagId, entry.path("greeting").asText());
+ var dag =
+ new DagDef(dagId)
+ .config("catchup", false)
+ .config("tags", List.of("java-sdk", "java_dynamic"));
+ dag.task("greet", Greet.class).config("queue", QUEUE);
+ dags.add(dag);
+ }
+ return dags;
+ }
+}
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/dynamic_dags.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/dynamic_dags.py
new file mode 100644
index 00000000000..4af37cecc71
--- /dev/null
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/e2e_test_utils/dynamic_dags.py
@@ -0,0 +1,71 @@
+# 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.
+"""Helpers for the e2e tests of language SDK Dags generated from a
``dags.yaml`` in the Dag bundle.
+
+Each entry of the YAML has a ``dag_id`` and a ``greeting``, and its Dag has
one ``greet`` task that
+returns the greeting.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+import yaml
+
+if TYPE_CHECKING:
+ from pathlib import Path
+
+ from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
+
+
+def write_yaml_atomically(path: Path, greetings: dict[str, str]) -> None:
+ """Replace the YAML in one step, so the Dag processor never reads half of
it."""
+ entries = [{"dag_id": dag_id, "greeting": greeting} for dag_id, greeting
in greetings.items()]
+ tmp_path = path.with_name(f"{path.name}.tmp")
+ tmp_path.write_text(yaml.safe_dump({"dags": entries}))
+ os.replace(tmp_path, path)
+
+
+def swap_yaml_and_check(
+ client: AirflowClient,
+ yaml_path: Path,
+ greetings: dict[str, str],
+ *,
+ previous: set[str],
+ tags: list[str],
+ timeout: int,
+ task_timeout: int,
+) -> set[str]:
+ """Swap in a YAML of *greetings*, then check the Dags, a run of each, and
the Dags it dropped."""
+ write_yaml_atomically(yaml_path, greetings)
+ client.wait_for_dag_ids(set(greetings), tags=tags, timeout=timeout)
+
+ run_ids = {
+ dag_id: client.trigger_dag(dag_id, json={"logical_date":
datetime.now(UTC).isoformat()})["dag_run_id"]
+ for dag_id in greetings
+ }
+ for dag_id, run_id in run_ids.items():
+ state = client.wait_for_dag_run(dag_id=dag_id, run_id=run_id,
timeout=task_timeout)
+ assert state == "success", f"{dag_id} run {run_id} ended {state!r}"
+ xcom = client.get_xcom_value(dag_id=dag_id, task_id="greet",
run_id=run_id, key="return_value")
+ assert xcom["value"] == greetings[dag_id], f"{dag_id} returned
{xcom['value']!r}"
+
+ for dag_id in previous - set(greetings):
+ assert client.get_dag(dag_id)["is_stale"], f"{dag_id} left the YAML
but is not stale"
+ return set(greetings)
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_dynamic_dag.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_dynamic_dag.py
index ca9c3e07269..a5dcf0ab894 100644
---
a/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_dynamic_dag.py
+++
b/airflow-e2e-tests/tests/airflow_e2e_tests/go_sdk_tests/test_go_sdk_dynamic_dag.py
@@ -28,13 +28,10 @@ with no rebuild, and checks that the live Dag IDs follow it
and that each live D
from __future__ import annotations
-import os
-from datetime import UTC, datetime
from pathlib import Path
-import yaml
-
from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
+from airflow_e2e_tests.e2e_test_utils.dynamic_dags import swap_yaml_and_check
# The Dag processor execs the binary to parse it, so the Dags appear some time
after the stack is up.
_PARSE_TIMEOUT = 600
@@ -46,48 +43,20 @@ _GO_TASK_TIMEOUT = 600
_TAGS = ["go_dynamic"]
-def _write_yaml_atomically(path: Path, greetings: dict[str, str]) -> None:
- """Replace the YAML in one step, so the Dag processor never reads half of
it."""
- entries = [{"dag_id": dag_id, "greeting": greeting} for dag_id, greeting
in greetings.items()]
- tmp_path = path.with_name(f"{path.name}.tmp")
- tmp_path.write_text(yaml.safe_dump({"dags": entries}))
- os.replace(tmp_path, path)
-
-
-def _swap_yaml_and_check(
- client: AirflowClient, yaml_path: Path, greetings: dict[str, str],
previous: set[str], timeout: int
-) -> set[str]:
- """Swap in a YAML of *greetings*, then check the Dags, a run of each, and
the Dags it dropped."""
- _write_yaml_atomically(yaml_path, greetings)
- client.wait_for_dag_ids(set(greetings), tags=_TAGS, timeout=timeout)
-
- run_ids = {
- dag_id: client.trigger_dag(dag_id, json={"logical_date":
datetime.now(UTC).isoformat()})["dag_run_id"]
- for dag_id in greetings
- }
- for dag_id, run_id in run_ids.items():
- state = client.wait_for_dag_run(dag_id=dag_id, run_id=run_id,
timeout=_GO_TASK_TIMEOUT)
- assert state == "success", f"{dag_id} run {run_id} ended {state!r}"
- xcom = client.get_xcom_value(dag_id=dag_id, task_id="greet",
run_id=run_id, key="return_value")
- assert xcom["value"] == greetings[dag_id], f"{dag_id} returned
{xcom['value']!r}"
-
- for dag_id in previous - set(greetings):
- assert client.get_dag(dag_id)["is_stale"], f"{dag_id} left the YAML
but is not stale"
- return set(greetings)
-
-
def test_swapping_the_yaml_changes_the_live_dags(airflow_dags_path: Path):
client = AirflowClient()
yaml_path = airflow_dags_path / "go-dynamic" / "dags.yaml"
- live = _swap_yaml_and_check(
+ live = swap_yaml_and_check(
client,
yaml_path,
{"go_dynamic_a": "hello from a", "go_dynamic_b": "hello from b",
"go_dynamic_c": "hello from c"},
previous=set(),
+ tags=_TAGS,
timeout=_PARSE_TIMEOUT,
+ task_timeout=_GO_TASK_TIMEOUT,
)
- live = _swap_yaml_and_check(
+ live = swap_yaml_and_check(
client,
yaml_path,
{
@@ -98,13 +67,17 @@ def
test_swapping_the_yaml_changes_the_live_dags(airflow_dags_path: Path):
"go_dynamic_e": "hello from e",
},
previous=live,
+ tags=_TAGS,
timeout=_RELOAD_TIMEOUT,
+ task_timeout=_GO_TASK_TIMEOUT,
)
# a keeps its Dag ID but gets a new greeting, so its run shows the Dag was
rebuilt from the new YAML.
- _swap_yaml_and_check(
+ swap_yaml_and_check(
client,
yaml_path,
{"go_dynamic_a": "hello again from a", "go_dynamic_f": "hello from f"},
previous=live,
+ tags=_TAGS,
timeout=_RELOAD_TIMEOUT,
+ task_timeout=_GO_TASK_TIMEOUT,
)
diff --git
a/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dynamic_dag.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dynamic_dag.py
new file mode 100644
index 00000000000..d22e48c55d4
--- /dev/null
+++
b/airflow-e2e-tests/tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dynamic_dag.py
@@ -0,0 +1,89 @@
+# 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.
+"""E2E test for the native Java Dags that ``java-native-bundle`` generates
from a YAML file.
+
+Run with::
+
+ E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \\
+ tests/airflow_e2e_tests/java_sdk_tests/test_java_sdk_dynamic_dag.py
-xvs
+
+``conftest._setup_java_sdk_integration`` puts the bundle JAR into the Dags
folder, with no ``dags.yaml``
+next to it yet. The JAR generates one Dag for each entry of the YAML the test
writes there, so the test
+swaps only the YAML, with no rebuild, and checks that the live Dag IDs follow
it and that each live Dag
+runs.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient
+from airflow_e2e_tests.e2e_test_utils.dynamic_dags import swap_yaml_and_check
+
+# The Dag processor starts a JVM to parse the JAR, so the Dags appear some
time after the stack is up.
+_PARSE_TIMEOUT = 600
+# After that, the Dag processor re-parses every few seconds (see java.yml),
and a Dag that left the
+# YAML goes stale a few seconds after that.
+_RELOAD_TIMEOUT = 180
+_JAVA_TASK_TIMEOUT = 600
+
+_TAGS = ["java_dynamic"]
+
+
+def test_swapping_the_yaml_changes_the_live_dags(airflow_dags_path: Path):
+ client = AirflowClient()
+ yaml_path = airflow_dags_path / "java-native" / "dags.yaml"
+
+ live = swap_yaml_and_check(
+ client,
+ yaml_path,
+ {
+ "java_dynamic_a": "hello from a",
+ "java_dynamic_b": "hello from b",
+ "java_dynamic_c": "hello from c",
+ },
+ previous=set(),
+ tags=_TAGS,
+ timeout=_PARSE_TIMEOUT,
+ task_timeout=_JAVA_TASK_TIMEOUT,
+ )
+ live = swap_yaml_and_check(
+ client,
+ yaml_path,
+ {
+ "java_dynamic_a": "hello from a",
+ "java_dynamic_b": "hello from b",
+ "java_dynamic_c": "hello from c",
+ "java_dynamic_d": "hello from d",
+ "java_dynamic_e": "hello from e",
+ },
+ previous=live,
+ tags=_TAGS,
+ timeout=_RELOAD_TIMEOUT,
+ task_timeout=_JAVA_TASK_TIMEOUT,
+ )
+ # a keeps its Dag ID but gets a new greeting. The greeting is not in the
serialized Dag, so its run
+ # shows that the worker built the Dag from the new YAML, not that the Dag
processor rebuilt it.
+ swap_yaml_and_check(
+ client,
+ yaml_path,
+ {"java_dynamic_a": "hello again from a", "java_dynamic_f": "hello from
f"},
+ previous=live,
+ tags=_TAGS,
+ timeout=_RELOAD_TIMEOUT,
+ task_timeout=_JAVA_TASK_TIMEOUT,
+ )
