This is an automated email from the ASF dual-hosted git repository.
uranusjr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 77c29134b37 Load Java Dags from a bundle path that holds a colon
(#76432)
77c29134b37 is described below
commit 77c29134b3786828c4450553de0e4870c69a02f7
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Nov 4 23:18:56 2026 +0800
Load Java Dags from a bundle path that holds a colon (#76432)
Co-authored-by: Claude <[email protected]>
Co-authored-by: Tzu-ping Chung <[email protected]>
---
.../language-sdks/java.rst | 21 ++
.../airflow/sdk/coordinators/java/coordinator.py | 95 +++++++-
.../task_sdk/coordinators/java/test_coordinator.py | 240 +++++++++++++++++++++
3 files changed, 354 insertions(+), 2 deletions(-)
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
index 2e8aba27521..621864a40f0 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
@@ -1092,6 +1092,27 @@ Register every Dag bundle a coordinator names in
``[dag_processor] dag_bundle_co
The ``[sdk]`` configuration is read at startup, so restart these components
after changing it. A rebuilt JAR is
picked up on the next task, since every task instance starts a new JVM.
+.. _java-sdk/bundle-path-separator:
+
+Dag bundle paths that contain a colon
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The ``java`` command cannot load a classpath containing paths that use the
``:`` (colon) character (on POSIX
+systems). The version directory of a Dag bundle may contain one, for example
when it is named after a
+timestamp. The coordinator tries to handle this automatically by reading such
a bundle through a symbolic link
+created in a temporary directory. A file or directory *inside* the Dag bundle
still cannot contain the colon
+separator.
+
+This approach should generally work. On the off chance it does not, however,
all such links are located in
+``airflow-java-classpath-<uid>`` under the temporary directory (``$TMPDIR`` on
Linux, usually ``/tmp``).
+Airflow logs should also report which link corresponds to which bundle.
+
+Each link is named after the first 16 hexadecimal digits of the SHA-256 hash
of the bundle version's absolute
+path, so the same path always gets the same link. Links are not removed
immediately when they are not needed,
+so some dead ones are expected. They are harmless. Airflow does delayed
cleanups, so they should not grow
+unbounded. You can also safely delete them *when Airflow is not running*;
Airflow will recreate them
+automatically the next time they are needed.
+
.. _java-sdk/coordinator-config:
:class:`~airflow.sdk.coordinators.java.JavaCoordinator` configuration
diff --git a/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
b/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
index d2888f6891c..18b6b75923b 100644
--- a/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
+++ b/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
@@ -19,9 +19,13 @@
from __future__ import annotations
+import contextlib
+import hashlib
import os
import pathlib
+import secrets
import stat
+import tempfile
import zipfile
import zlib
from typing import TYPE_CHECKING, Final
@@ -92,13 +96,95 @@ def _iter_dir(directory: pathlib.Path) ->
Iterator[pathlib.Path]:
yield from children
+def _get_classpath_link_dir() -> pathlib.Path:
+ """
+ Return this user's directory of classpath links, creating it if needed.
+
+ :raises ValueError: when the directory's path holds ``os.pathsep``.
+ :raises PermissionError: when the directory is not a real directory owned
by this user, or
+ others can write to it. Anyone who can write there could point a link
at their own JARs.
+ """
+ uid = os.getuid()
+ path = pathlib.Path(tempfile.gettempdir(), f"airflow-java-classpath-{uid}")
+ if os.pathsep in os.fspath(path):
+ raise ValueError(
+ f"Cannot keep Java classpath links in {path}, because its path
holds {os.pathsep!r}. "
+ "Point TMPDIR to a directory whose path does not."
+ )
+ with contextlib.suppress(FileExistsError):
+ path.mkdir(mode=0o700)
+ st = path.lstat()
+ if not stat.S_ISDIR(st.st_mode) or st.st_uid != uid or st.st_mode & 0o022:
+ raise PermissionError(
+ f"Cannot keep Java classpath links in {path}: it must be a
directory owned by "
+ "the current user that no other user can write to."
+ )
+ return path
+
+
+def _remove_dangling_links(link_dir: pathlib.Path) -> None:
+ for entry in link_dir.iterdir():
+ if entry.is_symlink() and not entry.exists():
+ with contextlib.suppress(FileNotFoundError):
+ entry.unlink()
+
+
+def _get_classpath_link(root: pathlib.Path) -> pathlib.Path:
+ """
+ Return a symlink to *root* whose own path holds no ``os.pathsep``.
+
+ The JVM splits a classpath on ``os.pathsep`` and has no way to escape it.
The link name is a
+ hash of *root*, so every process of this user shares one link per root. It
is replaced
+ atomically, and a link whose target is gone is removed when a new one is
created.
+ """
+ target = os.fspath(root.absolute())
+ link_dir = _get_classpath_link_dir()
+ link = link_dir / hashlib.sha256(os.fsencode(target)).hexdigest()[:16]
+ with contextlib.suppress(OSError):
+ if os.readlink(link) == target:
+ return link
+ _remove_dangling_links(link_dir)
+ staging = link_dir /
f"{link.name}.{os.getpid()}.{secrets.token_hex(4)}.tmp"
+ os.symlink(target, staging, target_is_directory=True)
+ try:
+ os.replace(staging, link)
+ except BaseException:
+ staging.unlink(missing_ok=True)
+ raise
+ log.debug("Linked a bundle path that holds the classpath separator",
bundle_path=target, link=link)
+ return link
+
+
+def _find_root(path: pathlib.Path, roots: Sequence[pathlib.Path]) ->
pathlib.Path | None:
+ return next((root for root in roots if path.is_relative_to(root)), None)
+
+
def _calculate_classpath(roots: Sequence[pathlib.Path], *, primary:
pathlib.Path | None = None) -> str:
- """Return the classpath of every JAR under *roots*. *primary*, when given,
goes first."""
+ """
+ Return the classpath of every JAR under *roots*. *primary*, when given,
goes first.
+
+ A root whose path holds ``os.pathsep``, such as a bundle version directory
named after a
+ timestamp, is reached through a symlink from :func:`_get_classpath_link`,
which this creates
+ if needed.
+
+ :raises ValueError: when a JAR's path below its root still holds
``os.pathsep``, or when the
+ temporary directory's path holds it.
+ """
jars = sorted(_find_jars(roots), key=lambda p: p.as_posix())
if primary is not None:
target = primary.resolve()
jars = [primary, *(p for p in jars if p.resolve() != target)]
- return os.pathsep.join(p.as_posix() for p in jars)
+ placed = [(p, _find_root(p, roots)) for p in jars]
+ if bad := [p for p, r in placed if os.pathsep in os.fspath(p if r is None
else p.relative_to(r))]:
+ raise ValueError(
+ f"Cannot put these JARs on a Java classpath, because
{os.pathsep!r} separates its "
+ f"entries: {', '.join(map(os.fspath, bad))}. Rename or move them."
+ )
+ linked_roots = {r for _, r in placed if r is not None and os.pathsep in
os.fspath(r)}
+ links = {r: _get_classpath_link(r) for r in linked_roots}
+ for root, link in links.items():
+ log.info("Bundle reached by classpath through a link",
bundle_path=root, link=link)
+ return os.pathsep.join((links[r] / p.relative_to(r) if r in links else
p).as_posix() for p, r in placed)
@attrs.define
@@ -262,6 +348,11 @@ class JavaCoordinator(SubprocessCoordinator):
picks the one that parses a bundle. A JAR whose Main-Class another JAR in
the
bundle also sets is rejected.
+ The JVM cannot load a JAR whose path holds the classpath separator (``:``
on
+ POSIX). The coordinator reaches a Dag bundle whose own path holds it
through
+ a symlink in the temporary directory, but a file or directory name inside
+ the bundle must not hold it.
+
The default *task_startup_timeout* should plenty long enough since a task-
containing JAR is not supposed to consume significant time to perform setup
(it should happen in individual tasks instead). However, if the launch time
diff --git a/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py
b/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py
index 15ebfb75426..f79406aa45b 100644
--- a/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py
+++ b/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py
@@ -18,11 +18,14 @@
from __future__ import annotations
+import concurrent.futures
import json
import os
import pathlib
import re
+import shutil
import socket
+import stat
import subprocess
import zlib
from unittest.mock import MagicMock, patch
@@ -36,6 +39,7 @@ from airflow.sdk.coordinators._subprocess import
_PopenActivitySubprocess
from airflow.sdk.coordinators.java.coordinator import (
JavaCoordinator,
_calculate_classpath,
+ _get_classpath_link,
_JarInfo,
_JarMetadata,
_walk_jars,
@@ -167,6 +171,201 @@ class TestCalculateClasspath:
assert result.split(os.pathsep) == [b.as_posix(), a.as_posix()]
+TIMESTAMPED_VERSION = f"2026-11-02T15{os.pathsep}52{os.pathsep}20Z"
+
+
[email protected]
+def link_dir(tmp_path):
+ """Point the coordinator's temporary directory into *tmp_path* and return
where its links go."""
+ tmp = tmp_path / "tmp"
+ tmp.mkdir()
+ with patch(
+ "airflow.sdk.coordinators.java.coordinator.tempfile.gettempdir",
+ autospec=True,
+ return_value=os.fspath(tmp),
+ ):
+ yield tmp / f"airflow-java-classpath-{os.getuid()}"
+
+
+def _make_timestamped_bundle(parent: pathlib.Path, name: str =
TIMESTAMPED_VERSION) -> pathlib.Path:
+ root = parent / name
+ (root / "libs").mkdir(parents=True)
+ _make_jar(root / "app.jar", main_class="com.example.Dags",
schema_version="2026-10-30")
+ _make_jar(root / "libs" / "dep.jar", main_class=None)
+ return root
+
+
+class TestClasspathLink:
+ def test_root_without_separator_is_not_linked(self, tmp_path, link_dir):
+ jar = _make_jar(tmp_path / "app.jar")
+
+ assert _calculate_classpath([tmp_path]) == jar.as_posix()
+ assert not link_dir.exists()
+
+ def test_root_with_separator_is_reached_through_a_link(self, tmp_path,
link_dir):
+ root = _make_timestamped_bundle(tmp_path)
+
+ result = _calculate_classpath([root], primary=root / "libs" /
"dep.jar")
+
+ [link] = link_dir.iterdir()
+ assert result.split(os.pathsep) == [f"{link.as_posix()}/libs/dep.jar",
f"{link.as_posix()}/app.jar"]
+ assert os.readlink(link) == os.fspath(root)
+ assert stat.S_IMODE(link_dir.stat().st_mode) == 0o700
+
+ def test_mixed_roots_link_only_the_root_with_separator(self, tmp_path,
link_dir):
+ plain = tmp_path / "plain"
+ plain.mkdir()
+ lib = _make_jar(plain / "lib.jar", main_class=None)
+ root = _make_timestamped_bundle(tmp_path)
+
+ result = _calculate_classpath([plain, root])
+
+ [link] = link_dir.iterdir()
+ assert os.readlink(link) == os.fspath(root)
+ assert result.split(os.pathsep) == [
+ f"{link.as_posix()}/app.jar",
+ f"{link.as_posix()}/libs/dep.jar",
+ lib.as_posix(),
+ ]
+
+ @pytest.mark.parametrize(
+ ("inner_first", "link_count"),
+ [pytest.param(False, 1, id="outer-first"), pytest.param(True, 2,
id="inner-first")],
+ )
+ def test_nested_roots(self, tmp_path, link_dir, inner_first, link_count):
+ root = _make_timestamped_bundle(tmp_path)
+ roots = [root / "libs", root] if inner_first else [root, root / "libs"]
+
+ entries = _calculate_classpath(roots).split(os.pathsep)
+
+ assert sorted(pathlib.Path(e).resolve() for e in entries) == sorted(
+ [(root / "app.jar").resolve(), (root / "libs" /
"dep.jar").resolve()]
+ )
+ assert len(list(link_dir.iterdir())) == link_count
+
+ def test_link_is_reused(self, tmp_path, link_dir):
+ root = _make_timestamped_bundle(tmp_path)
+ link = _get_classpath_link(root)
+ inode = link.lstat().st_ino
+
+ assert _get_classpath_link(root) == link
+ assert link.lstat().st_ino == inode
+
+ @patch(
+ "airflow.sdk.coordinators.java.coordinator.os.replace",
+ autospec=True,
+ side_effect=OSError("replace failed"),
+ )
+ def test_staging_link_is_removed_when_replacing_fails(self, mock_replace,
tmp_path, link_dir):
+ root = _make_timestamped_bundle(tmp_path)
+
+ with pytest.raises(OSError, match="replace failed"):
+ _get_classpath_link(root)
+
+ assert list(link_dir.iterdir()) == []
+
+ def test_link_to_another_target_is_replaced(self, tmp_path, link_dir):
+ root = _make_timestamped_bundle(tmp_path)
+ link = _get_classpath_link(root)
+ link.unlink()
+ link.symlink_to(tmp_path)
+
+ assert _get_classpath_link(root) == link
+ assert os.readlink(link) == os.fspath(root)
+
+ def test_dangling_links_are_removed_when_a_link_is_created(self, tmp_path,
link_dir):
+ old = _get_classpath_link(_make_timestamped_bundle(tmp_path / "old"))
+ kept = _get_classpath_link(_make_timestamped_bundle(tmp_path / "kept"))
+ shutil.rmtree(tmp_path / "old")
+
+ new = _get_classpath_link(_make_timestamped_bundle(tmp_path / "new"))
+
+ assert sorted(link_dir.iterdir()) == sorted([kept, new])
+ assert not old.is_symlink()
+
+ def test_concurrent_callers_share_one_link(self, tmp_path, link_dir):
+ root = _make_timestamped_bundle(tmp_path)
+
+ with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
+ links = set(pool.map(lambda _: _get_classpath_link(root),
range(64)))
+
+ assert len(links) == 1
+ assert list(link_dir.iterdir()) == [*links]
+
+ @pytest.mark.parametrize("unsafe", ["writable-by-others", "symlink",
"file", "other-owner"])
+ def test_rejects_an_unsafe_link_dir(self, tmp_path, link_dir, unsafe):
+ root = _make_timestamped_bundle(tmp_path)
+ uid = os.getuid()
+ if unsafe == "writable-by-others":
+ link_dir.mkdir(mode=0o700)
+ link_dir.chmod(0o777)
+ elif unsafe == "symlink":
+ (tmp_path / "elsewhere").mkdir(mode=0o700)
+ link_dir.symlink_to(tmp_path / "elsewhere")
+ elif unsafe == "file":
+ link_dir.write_bytes(b"")
+ else:
+ uid += 1
+
+ with (
+ patch("airflow.sdk.coordinators.java.coordinator.os.getuid",
autospec=True, return_value=uid),
+ pytest.raises(PermissionError, match="Cannot keep Java classpath
links"),
+ ):
+ _get_classpath_link(root)
+
+ def test_rejects_a_link_dir_holding_the_separator(self, tmp_path):
+ root = _make_timestamped_bundle(tmp_path)
+ tmp = tmp_path / f"tmp{os.pathsep}dir"
+ tmp.mkdir()
+
+ with (
+ patch(
+
"airflow.sdk.coordinators.java.coordinator.tempfile.gettempdir",
+ autospec=True,
+ return_value=os.fspath(tmp),
+ ),
+ pytest.raises(ValueError, match="Cannot keep Java classpath
links"),
+ ):
+ _calculate_classpath([root])
+
+ assert list(tmp.iterdir()) == []
+
+ def test_separator_inside_the_bundle_raises_without_linking(self,
tmp_path, link_dir):
+ root = _make_timestamped_bundle(tmp_path)
+ nested = root / f"a{os.pathsep}b"
+ nested.mkdir()
+ jar = _make_jar(nested / "app.jar")
+
+ with pytest.raises(ValueError, match="Cannot put these JARs on a Java
classpath") as ctx:
+ _calculate_classpath([root])
+
+ assert os.fspath(jar) in str(ctx.value)
+ assert not link_dir.exists()
+
+ @pytest.mark.skipif(shutil.which("java") is None or shutil.which("javac")
is None, reason="needs a JDK")
+ def test_jvm_loads_jars_through_the_link(self, tmp_path, link_dir):
+ src = tmp_path / "src"
+ src.mkdir()
+ (src / "Dep.java").write_text('public class Dep { public static String
hi() { return "loaded"; } }')
+ (src / "Main.java").write_text(
+ "public class Main { public static void main(String[] a) {
System.out.println(Dep.hi()); } }"
+ )
+ subprocess.run(["javac", "-d", os.fspath(src), *map(os.fspath,
src.glob("*.java"))], check=True)
+ root = tmp_path / TIMESTAMPED_VERSION
+ (root / "libs").mkdir(parents=True)
+ make_jar(root / "app.jar", entries={"Main.class": (src /
"Main.class").read_bytes()})
+ make_jar(root / "libs" / "dep.jar", entries={"Dep.class": (src /
"Dep.class").read_bytes()})
+
+ result = subprocess.run(
+ ["java", "-classpath", _calculate_classpath([root]), "Main"],
+ capture_output=True,
+ text=True,
+ check=True,
+ )
+
+ assert result.stdout.strip() == "loaded"
+
+
class TestMainJar:
def test_returns_main_class_from_jar(self, tmp_path):
_make_jar(tmp_path.joinpath("app.jar"), main_class="com.example.Main",
schema_version="2026-06-16")
@@ -408,6 +607,23 @@ class TestJavaCoordinatorExecuteTask:
classpath = cmd[cp_idx + 1]
assert jars_dir.joinpath("app.jar").as_posix() in classpath
+ def test_timestamped_bundle_path_is_linked(self, tmp_path, link_dir,
mock_client):
+ root = _make_timestamped_bundle(tmp_path)
+ bundle = {
+ "name": "java-task-handlers",
+ "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+ "kwargs": {"path": str(root)},
+ }
+ with conf_vars({("dag_processor", "dag_bundle_config_list"):
json.dumps([bundle])}):
+ cmd = self._captured_popen_cmd(bundle["name"], mock_client)
+
+ [link] = link_dir.iterdir()
+ classpath = cmd[cmd.index("-classpath") + 1]
+ assert classpath.split(os.pathsep) == [
+ f"{link.as_posix()}/app.jar",
+ f"{link.as_posix()}/libs/dep.jar",
+ ]
+
def test_main_class_present(self, java_task_handlers, mock_client):
cmd = self._captured_popen_cmd(java_task_handlers, mock_client)
assert "com.example.TaskRunner" in cmd
@@ -707,3 +923,27 @@ class TestBuildParseDagCommand:
"--comm=127.0.0.1:1001",
"--logs=127.0.0.1:1002",
]
+
+
@patch("airflow.sdk.coordinators._subprocess._set_close_on_exec_above_stderr",
autospec=True)
+ @patch("airflow.sdk.coordinators._subprocess.signal.signal", autospec=True)
+ @patch("airflow.sdk.coordinators._subprocess.os.execvpe", autospec=True,
side_effect=OSError("exec"))
+ def test_parse_dag_links_a_timestamped_bundle_path(
+ self, mock_execvpe, mock_signal, mock_close_on_exec, tmp_path, link_dir
+ ):
+ root = _make_timestamped_bundle(tmp_path)
+
+ with pytest.raises(OSError, match="exec"):
+ JavaCoordinator().parse_dag(
+ path=root / "app.jar",
+ bundle_path=root,
+ comm_address=("127.0.0.1", 1001),
+ logs_address=("127.0.0.1", 1002),
+ report_schema_version=lambda _: None,
+ )
+
+ [link] = link_dir.iterdir()
+ classpath = mock_execvpe.call_args.args[1][2]
+ assert classpath.split(os.pathsep) == [
+ f"{link.as_posix()}/app.jar",
+ f"{link.as_posix()}/libs/dep.jar",
+ ]
