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

This reverts commit 51140809d2.

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

@@ -451,7 +451,6 @@ 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()
@@ -473,7 +472,6 @@ 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().
@@ -738,14 +736,6 @@ 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*.
@@ -811,7 +801,6 @@ 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 = [
@@ -826,8 +815,6 @@ 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
@@ -949,7 +936,6 @@ 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(
@@ -963,7 +949,6 @@ 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,30 +92,13 @@ def _iter_dir(directory: pathlib.Path) -> Iterator[pathlib.Path]:
     yield from children
 
 
-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``.
-    """
+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."""
     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)]
-    # "./" 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)
+    return os.pathsep.join(p.as_posix() for p in jars)
 
 
 @attrs.define
@@ -297,15 +280,11 @@ class JavaCoordinator(SubprocessCoordinator):
         return [
             self.java_executable,
             "-classpath",
-            _calculate_classpath(roots, cwd=self._get_working_directory(), primary=primary),
+            _calculate_classpath(roots, 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,26 +127,33 @@ class TestJarMetadata:
 
 class TestCalculateClasspath:
     def test_single_jar(self, tmp_path):
-        tmp_path.joinpath("app.jar").write_bytes(b"")
-        result = _calculate_classpath([tmp_path], cwd=tmp_path)
-        assert result == "./app.jar"
+        jar = tmp_path.joinpath("app.jar")
+        jar.write_bytes(b"")
+        result = _calculate_classpath([tmp_path])
+        assert result == jar.as_posix()
 
     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], cwd=tmp_path)
-        assert set(result.split(os.pathsep)) == {"./a.jar", "./b.jar", "./c.jar"}
+        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(),
+        }
 
     def test_non_jar_files_excluded(self, tmp_path):
-        tmp_path.joinpath("app.jar").write_bytes(b"")
+        jar = tmp_path.joinpath("app.jar")
+        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], cwd=tmp_path)
-        assert result == "./app.jar"
+        result = _calculate_classpath([tmp_path])
+        assert result == jar.as_posix()
 
     def test_empty_directory_returns_empty_string(self, tmp_path):
-        result = _calculate_classpath([tmp_path], cwd=tmp_path)
+        result = _calculate_classpath([tmp_path])
         assert result == ""
 
     def test_primary_jar_goes_first(self, tmp_path):
@@ -155,35 +162,9 @@ class TestCalculateClasspath:
         b = tmp_path.joinpath("b.jar")
         b.write_bytes(b"")
 
-        result = _calculate_classpath([tmp_path], cwd=tmp_path, primary=b)
+        result = _calculate_classpath([tmp_path], primary=b)
 
-        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)
+        assert result.split(os.pathsep) == [b.as_posix(), a.as_posix()]
 
 
 class TestMainJar:
@@ -334,16 +315,10 @@ 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(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
+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
 
 
 @pytest.fixture
@@ -390,7 +365,6 @@ class TestJavaCoordinatorExecuteTask:
 
         def capture_popen(cmd, **kwargs):
             popen_calls.append(cmd)
-            self.popen_kwargs = kwargs
             return mock_proc
 
         with (
@@ -432,13 +406,7 @@ class TestJavaCoordinatorExecuteTask:
         assert "-classpath" in cmd
         cp_idx = cmd.index("-classpath")
         classpath = cmd[cp_idx + 1]
-        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"
+        assert jars_dir.joinpath("app.jar").as_posix() in classpath
 
     def test_main_class_present(self, java_task_handlers, mock_client):
         cmd = self._captured_popen_cmd(java_task_handlers, mock_client)
@@ -618,17 +586,17 @@ class TestBuildParseDagCommand:
 
         command, schema_version = self._build(coordinator, tmp_path, jar)
 
-        assert command == ["/opt/java/bin/java", "-classpath", "./app.jar", "-Xmx256m", "com.example.Dags"]
+        assert command == ["/opt/java/bin/java", "-classpath", jar.as_posix(), "-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()
-        _make_jar(tmp_path / "libs" / "airflow-sdk.jar", main_class=None, schema_version="2026-10-30")
+        sdk = _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) == [f"./{jar.name}", "./libs/airflow-sdk.jar"]
+        assert command[2].split(os.pathsep) == [jar.as_posix(), sdk.as_posix()]
         assert command[-1] == "com.example.Dags"
         assert schema_version == "2026-10-30"
 
@@ -639,7 +607,7 @@ class TestBuildParseDagCommand:
 
         command, _ = self._build(JavaCoordinator(), tmp_path, jar)
 
-        assert command[2].split(os.pathsep) == [f"./{jar.name}", f"./{sibling.name}"]
+        assert command[2].split(os.pathsep) == [jar.as_posix(), sibling.as_posix()]
 
     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")
@@ -715,21 +683,15 @@ 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"))
-    @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")
+    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")
         reported: list[str | None] = []
 
         with pytest.raises(OSError, match="exec"):
             JavaCoordinator().parse_dag(
                 path=jar,
-                bundle_path=root,
+                bundle_path=tmp_path,
                 comm_address=("127.0.0.1", 1001),
                 logs_address=("127.0.0.1", 1002),
                 report_schema_version=reported.append,
@@ -737,11 +699,10 @@ 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",
-            "./app.jar",
+            jar.as_posix(),
             "com.example.Dags",
             "--comm=127.0.0.1:1001",
             "--logs=127.0.0.1:1002",