This is an automated email from the ASF dual-hosted git repository. vatsrahul1001 pushed a commit to branch revert-76314 in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 23228327846db87d3c9bd7a2ec3280f90c9cc000 Author: Rahul Vats <[email protected]> AuthorDate: Tue Nov 3 11:57:11 2026 +0530 Revert "Load Java Dags from a bundle path that contains a colon (#76314)" This reverts commit 51140809d2a7299955bdba51588947eeb6f015c1. --- .../src/airflow/sdk/coordinators/_subprocess.py | 15 ---- .../airflow/sdk/coordinators/java/coordinator.py | 29 +------ .../task_sdk/coordinators/java/test_coordinator.py | 99 +++++++--------------- 3 files changed, 34 insertions(+), 109 deletions(-) diff --git a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py b/task-sdk/src/airflow/sdk/coordinators/_subprocess.py index 706a51e7df5..7f69817c3ca 100644 --- a/task-sdk/src/airflow/sdk/coordinators/_subprocess.py +++ b/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) diff --git a/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py b/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py index f67b55851ca..d2888f6891c 100644 --- a/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py +++ b/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 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 0d086200e2c..36e7616089b 100644 --- a/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py +++ b/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",
