Load Java Dags from a bundle path that contains a colon (#76314)

* Load Java Dags from a bundle path that contains a colon

The JVM splits -classpath on the path separator (':' on Linux) and has
no way to escape it. When a Dag bundle's path contains one, for example
a version directory named after a timestamp like
2026-11-02T15:52:20Z, JavaCoordinator passed absolute JAR paths, the
JVM split them apart, never loaded the JAR and exited with code 1, so
the Dag failed to parse with "exited with code 1 without a parse
result".

JavaCoordinator now starts the JVM in the bundle root and puts the
JARs on the classpath relative to it, so the bundle's own path never
appears there. SubprocessCoordinator gets a _get_working_directory hook
that both the Dag parse (chdir before exec) and task execution (Popen
cwd) use. A JAR whose path inside the bundle still holds the separator
is rejected with an error that names it.

* Prefix relative Java classpath entries with ./

The java launcher reads an argument that starts with "@" as an @argfile,
including the value after -classpath, so a bundle whose first JAR sat
under a directory like "@scope" failed to start. Every entry now starts
with "./".

* Test a Java bundle path holding the path separator end to end

Run the task and Dag parse launch tests in a bundle directory named after
a timestamp too, so the working directory and the relative classpath are
checked together for both paths.

Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com>

---------

Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com>

task-sdk/src/airflow/sdk/coordinators/_subprocess.py

@@ -451,6 +451,7 @@ class _PopenActivitySubprocess(ActivitySubprocess):
         subprocess_schema_version: str | None = None,
         subprocess_logs_to_stdout: bool = False,
         startup_timeout: float = 10.0,
+        cwd: pathlib.Path | None = None,
         **kwargs,
     ) -> Self:
         process_log = logger or structlog.get_logger(logger_name="task").bind()
@@ -472,6 +473,7 @@ class _PopenActivitySubprocess(ActivitySubprocess):
                     stdout=stdout_w.fileno(),
                     stderr=stderr_w.fileno(),
                     env=_build_runtime_env(),
+                    cwd=cwd,
                     # Own process group, so the supervisor can signal everything the runtime
                     # spawned with os.killpg, as the Python task runner has done since #65505.
                     # Popen returns after exec, so the group exists before kill().
@@ -736,6 +738,14 @@ class SubprocessCoordinator(BaseCoordinator):
             )
         return self._active_scan_roots
 
+    def _get_working_directory(self) -> pathlib.Path | None:
+        """
+        Return the directory the runtime starts in, or ``None`` to keep the current one.
+
+        It is resolved after the command, while :meth:`_get_scan_roots` still answers.
+        """
+        return None
+
     def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
         """
         Build the subprocess command and resolve its supervisor wire-schema version for *what*.
@@ -801,6 +811,7 @@ class SubprocessCoordinator(BaseCoordinator):
         """
         with self._set_scan_roots([bundle_path]):
             command, schema_version = self._build_parse_dag_command(path=path)
+            cwd = self._get_working_directory()
         if schema_version is not None:
             get_schema_version_migrator().resolve_version(schema_version)
         argv = [
@@ -815,6 +826,8 @@ class SubprocessCoordinator(BaseCoordinator):
             if (sig := getattr(signal, name, None)) is not None:
                 signal.signal(sig, signal.SIG_DFL)
         _set_close_on_exec_above_stderr()
+        if cwd is not None:
+            os.chdir(cwd)
         os.execvpe(argv[0], argv, _build_runtime_env())
 
     @contextlib.contextmanager
@@ -936,6 +949,7 @@ class SubprocessCoordinator(BaseCoordinator):
                     )
                 else:
                     command, subprocess_schema_version = self._build_execute_task_command(what=what)
+                cwd = self._get_working_directory()
                 if subprocess_schema_version is not None:
                     _require_supported_schema_version(subprocess_schema_version)
             process = _PopenActivitySubprocess.start(
@@ -949,6 +963,7 @@ class SubprocessCoordinator(BaseCoordinator):
                 command=command,
                 subprocess_schema_version=subprocess_schema_version,
                 startup_timeout=self.task_startup_timeout,
+                cwd=cwd,
             )
             exit_code = process.wait()
             return self.ExecutionResult(exit_code, process.final_state)

task-sdk/src/airflow/sdk/coordinators/java/coordinator.py

@@ -92,13 +92,30 @@ def _iter_dir(directory: pathlib.Path) -> Iterator[pathlib.Path]:
     yield from children
 
 
-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."""
+def _calculate_classpath(
+    roots: Sequence[pathlib.Path], *, cwd: pathlib.Path, primary: pathlib.Path | None = None
+) -> str:
+    """
+    Return the classpath of every JAR under *roots*, relative to *cwd*. *primary*, when given, goes first.
+
+    The JVM splits a classpath on ``os.pathsep`` and has no way to escape it, so a JAR whose
+    absolute path holds one, such as a bundle version directory named after a timestamp, could
+    not be loaded. Relative to *cwd* that part of the path is gone.
+
+    :raises ValueError: when a path relative to *cwd* still holds ``os.pathsep``.
+    """
     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)
+    # "./" stops the launcher reading a classpath that starts with "@" as an @argfile.
+    entries = [f"./{pathlib.Path(os.path.relpath(p, cwd)).as_posix()}" for p in jars]
+    if bad := [e for e in entries if os.pathsep in e]:
+        raise ValueError(
+            f"Cannot put these JARs on a Java classpath, because {os.pathsep!r} separates its "
+            f"entries: {', '.join(bad)}. Rename them, or the directories under {os.fspath(cwd)} holding them."
+        )
+    return os.pathsep.join(entries)
 
 
 @attrs.define
@@ -280,11 +297,15 @@ class JavaCoordinator(SubprocessCoordinator):
         return [
             self.java_executable,
             "-classpath",
-            _calculate_classpath(roots, primary=primary),
+            _calculate_classpath(roots, cwd=self._get_working_directory(), primary=primary),
             *self.jvm_args,
             main_class,
         ]
 
+    def _get_working_directory(self) -> pathlib.Path:
+        # The classpath is relative to this, which keeps os.pathsep in the bundle's own path off it.
+        return self._get_scan_roots()[0]
+
     def _build_execute_task_command(self, *, what: TaskInstance) -> tuple[list[str], str | None]:
         # Without main_class, the first executable JAR in path order wins; tracked at
         # https://github.com/apache/airflow/issues/71134

task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py

@@ -127,33 +127,26 @@ class TestJarMetadata:
 
 class TestCalculateClasspath:
     def test_single_jar(self, tmp_path):
-        jar = tmp_path.joinpath("app.jar")
-        jar.write_bytes(b"")
-        result = _calculate_classpath([tmp_path])
-        assert result == jar.as_posix()
+        tmp_path.joinpath("app.jar").write_bytes(b"")
+        result = _calculate_classpath([tmp_path], cwd=tmp_path)
+        assert result == "./app.jar"
 
     def test_multiple_jars_all_included(self, tmp_path):
         tmp_path.joinpath("a.jar").write_bytes(b"")
         tmp_path.joinpath("b.jar").write_bytes(b"")
         tmp_path.joinpath("c.jar").write_bytes(b"")
-        result = _calculate_classpath([tmp_path])
-        entries = set(result.split(os.pathsep))
-        assert entries == {
-            tmp_path.joinpath("a.jar").as_posix(),
-            tmp_path.joinpath("b.jar").as_posix(),
-            tmp_path.joinpath("c.jar").as_posix(),
-        }
+        result = _calculate_classpath([tmp_path], cwd=tmp_path)
+        assert set(result.split(os.pathsep)) == {"./a.jar", "./b.jar", "./c.jar"}
 
     def test_non_jar_files_excluded(self, tmp_path):
-        jar = tmp_path.joinpath("app.jar")
-        jar.write_bytes(b"")
+        tmp_path.joinpath("app.jar").write_bytes(b"")
         tmp_path.joinpath("readme.txt").write_bytes(b"")
         tmp_path.joinpath("config.yaml").write_bytes(b"")
-        result = _calculate_classpath([tmp_path])
-        assert result == jar.as_posix()
+        result = _calculate_classpath([tmp_path], cwd=tmp_path)
+        assert result == "./app.jar"
 
     def test_empty_directory_returns_empty_string(self, tmp_path):
-        result = _calculate_classpath([tmp_path])
+        result = _calculate_classpath([tmp_path], cwd=tmp_path)
         assert result == ""
 
     def test_primary_jar_goes_first(self, tmp_path):
@@ -162,9 +155,35 @@ class TestCalculateClasspath:
         b = tmp_path.joinpath("b.jar")
         b.write_bytes(b"")
 
-        result = _calculate_classpath([tmp_path], primary=b)
+        result = _calculate_classpath([tmp_path], cwd=tmp_path, primary=b)
 
-        assert result.split(os.pathsep) == [b.as_posix(), a.as_posix()]
+        assert result.split(os.pathsep) == ["./b.jar", "./a.jar"]
+
+    def test_entry_starting_with_at_sign_is_not_an_argfile(self, tmp_path):
+        tmp_path.joinpath("@scope").mkdir()
+        tmp_path.joinpath("@scope", "dep.jar").write_bytes(b"")
+
+        result = _calculate_classpath([tmp_path], cwd=tmp_path)
+
+        assert result == "./@scope/dep.jar"
+
+    def test_path_separator_above_cwd_stays_off_the_classpath(self, tmp_path):
+        root = tmp_path / f"2026-11-02T15{os.pathsep}52{os.pathsep}20Z"
+        root.mkdir()
+        root.joinpath("app.jar").write_bytes(b"")
+        root.joinpath("dep.jar").write_bytes(b"")
+
+        result = _calculate_classpath([root], cwd=root, primary=root / "dep.jar")
+
+        assert result.split(os.pathsep) == ["./dep.jar", "./app.jar"]
+
+    def test_path_separator_below_cwd_raises(self, tmp_path):
+        nested = tmp_path / f"a{os.pathsep}b"
+        nested.mkdir()
+        nested.joinpath("app.jar").write_bytes(b"")
+
+        with pytest.raises(ValueError, match="Cannot put these JARs on a Java classpath"):
+            _calculate_classpath([tmp_path], cwd=tmp_path)
 
 
 class TestMainJar:
@@ -315,10 +334,16 @@ class TestJavaCoordinatorAttributes:
         assert schema_version == "2026-06-16"
 
 
+# A bundle version directory named after a timestamp holds os.pathsep.
+TIMESTAMPED_DIR = f"2026-11-02T15{os.pathsep}52{os.pathsep}20Z"
+
+
 @pytest.fixture
-def jars_dir(tmp_path):
-    _make_jar(tmp_path.joinpath("app.jar"), main_class="com.example.TaskRunner", schema_version="2026-06-16")
-    return tmp_path
+def jars_dir(request, tmp_path):
+    root = tmp_path / getattr(request, "param", "")
+    root.mkdir(exist_ok=True)
+    _make_jar(root / "app.jar", main_class="com.example.TaskRunner", schema_version="2026-06-16")
+    return root
 
 
 @pytest.fixture
@@ -365,6 +390,7 @@ class TestJavaCoordinatorExecuteTask:
 
         def capture_popen(cmd, **kwargs):
             popen_calls.append(cmd)
+            self.popen_kwargs = kwargs
             return mock_proc
 
         with (
@@ -406,7 +432,13 @@ class TestJavaCoordinatorExecuteTask:
         assert "-classpath" in cmd
         cp_idx = cmd.index("-classpath")
         classpath = cmd[cp_idx + 1]
-        assert jars_dir.joinpath("app.jar").as_posix() in classpath
+        assert classpath == "./app.jar"
+
+    @pytest.mark.parametrize("jars_dir", ["", TIMESTAMPED_DIR], indirect=True, ids=["plain", "timestamped"])
+    def test_runs_in_the_bundle_directory(self, jars_dir, java_task_handlers, mock_client):
+        cmd = self._captured_popen_cmd(java_task_handlers, mock_client)
+        assert self.popen_kwargs["cwd"] == jars_dir
+        assert cmd[cmd.index("-classpath") + 1] == "./app.jar"
 
     def test_main_class_present(self, java_task_handlers, mock_client):
         cmd = self._captured_popen_cmd(java_task_handlers, mock_client)
@@ -586,17 +618,17 @@ class TestBuildParseDagCommand:
 
         command, schema_version = self._build(coordinator, tmp_path, jar)
 
-        assert command == ["/opt/java/bin/java", "-classpath", jar.as_posix(), "-Xmx256m", "com.example.Dags"]
+        assert command == ["/opt/java/bin/java", "-classpath", "./app.jar", "-Xmx256m", "com.example.Dags"]
         assert schema_version == "2026-10-30"
 
     def test_thin_jar_takes_the_schema_version_from_a_sibling(self, tmp_path):
         jar = _make_jar(tmp_path / "app.jar", main_class="com.example.Dags")
         (tmp_path / "libs").mkdir()
-        sdk = _make_jar(tmp_path / "libs" / "airflow-sdk.jar", main_class=None, schema_version="2026-10-30")
+        _make_jar(tmp_path / "libs" / "airflow-sdk.jar", main_class=None, schema_version="2026-10-30")
 
         command, schema_version = self._build(JavaCoordinator(), tmp_path, jar)
 
-        assert command[2].split(os.pathsep) == [jar.as_posix(), sdk.as_posix()]
+        assert command[2].split(os.pathsep) == [f"./{jar.name}", "./libs/airflow-sdk.jar"]
         assert command[-1] == "com.example.Dags"
         assert schema_version == "2026-10-30"
 
@@ -607,7 +639,7 @@ class TestBuildParseDagCommand:
 
         command, _ = self._build(JavaCoordinator(), tmp_path, jar)
 
-        assert command[2].split(os.pathsep) == [jar.as_posix(), sibling.as_posix()]
+        assert command[2].split(os.pathsep) == [f"./{jar.name}", f"./{sibling.name}"]
 
     def test_matches_the_execute_command(self, tmp_path):
         jar = _make_jar(tmp_path / "app.jar", main_class="com.example.Dags", schema_version="2026-10-30")
@@ -683,15 +715,21 @@ class TestBuildParseDagCommand:
 
     @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.chdir", autospec=True)
     @patch("airflow.sdk.coordinators._subprocess.os.execvpe", autospec=True, side_effect=OSError("exec"))
-    def test_parse_dag_execs_the_jvm(self, mock_execvpe, mock_signal, mock_close_on_exec, tmp_path):
-        jar = _make_jar(tmp_path / "app.jar", main_class="com.example.Dags", schema_version="2026-10-30")
+    @pytest.mark.parametrize("bundle_dir", ["", TIMESTAMPED_DIR], ids=["plain", "timestamped"])
+    def test_parse_dag_execs_the_jvm(
+        self, mock_execvpe, mock_chdir, mock_signal, mock_close_on_exec, tmp_path, bundle_dir
+    ):
+        root = tmp_path / bundle_dir
+        root.mkdir(exist_ok=True)
+        jar = _make_jar(root / "app.jar", main_class="com.example.Dags", schema_version="2026-10-30")
         reported: list[str | None] = []
 
         with pytest.raises(OSError, match="exec"):
             JavaCoordinator().parse_dag(
                 path=jar,
-                bundle_path=tmp_path,
+                bundle_path=root,
                 comm_address=("127.0.0.1", 1001),
                 logs_address=("127.0.0.1", 1002),
                 report_schema_version=reported.append,
@@ -699,10 +737,11 @@ class TestBuildParseDagCommand:
 
         assert reported == ["2026-10-30"]
         argv = mock_execvpe.call_args.args[1]
+        mock_chdir.assert_called_once_with(root)
         assert argv == [
             "java",
             "-classpath",
-            jar.as_posix(),
+            "./app.jar",
             "com.example.Dags",
             "--comm=127.0.0.1:1001",
             "--logs=127.0.0.1:1002",