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 93a5e20f168 [v3-4-test] Don't retry Java tasks if classpath can't be
built (#76707) (#76719)
93a5e20f168 is described below
commit 93a5e20f168219c64192be4aced6295acc7dc3d4
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Nov 9 08:21:14 2026 +0000
[v3-4-test] Don't retry Java tasks if classpath can't be built (#76707)
(#76719)
A colon inside a Dag bundle, a temporary directory whose path holds one,
and an unsafe directory of classpath links now raise a non-retryable
TaskLaunchError instead of ValueError or PermissionError, which the
coordinator retried although a retry cannot fix them.
(cherry picked from commit e7ca5f1c2d383dccf35b7c38c59eb115f3dbdd7f)
Co-authored-by: Tzu-ping Chung <[email protected]>
---
.../language-sdks/java.rst | 6 ++---
.../airflow/sdk/coordinators/java/coordinator.py | 26 +++++++++++++---------
.../task_sdk/coordinators/java/test_coordinator.py | 10 ++++++---
3 files changed, 25 insertions(+), 17 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 0cc9981b52d..5039f0e6a23 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
@@ -1072,11 +1072,11 @@ The ``java`` command cannot load a classpath containing
paths that use the ``:``
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.
+separator. Such a Dag bundle fails to parse, and its tasks fail without
retrying.
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.
+``airflow-java-classpath-<uid>`` under the temporary directory (``/tmp`` by
default on Linux, or ``$TMPDIR``
+if set). 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,
diff --git a/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
b/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
index 18b6b75923b..7fdc6864276 100644
--- a/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
+++ b/task-sdk/src/airflow/sdk/coordinators/java/coordinator.py
@@ -40,6 +40,7 @@ from airflow.sdk.coordinators.java._jar_manifest import (
SUPERVISOR_SCHEMA_VERSION,
read_main_attributes,
)
+from airflow.sdk.execution_time.coordinator import TaskLaunchError
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator, Sequence
@@ -100,24 +101,25 @@ 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.
+ :raises TaskLaunchError: when the directory's path holds ``os.pathsep``, or
+ it is not a real directory owned by this user, or writable by others.
"""
uid = os.getuid()
path = pathlib.Path(tempfile.gettempdir(), f"airflow-java-classpath-{uid}")
if os.pathsep in os.fspath(path):
- raise ValueError(
+ raise TaskLaunchError(
f"Cannot keep Java classpath links in {path}, because its path
holds {os.pathsep!r}. "
- "Point TMPDIR to a directory whose path does not."
+ "Point TMPDIR to a directory whose path does not.",
+ retryable=False,
)
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(
+ raise TaskLaunchError(
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."
+ "the current user that no other user can write to.",
+ retryable=False,
)
return path
@@ -167,8 +169,8 @@ def _calculate_classpath(roots: Sequence[pathlib.Path], *,
primary: pathlib.Path
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.
+ :raises TaskLaunchError: when a JAR's path below its root still holds
``os.pathsep``, or the
+ directory of links cannot be used. Retrying cannot fix either.
"""
jars = sorted(_find_jars(roots), key=lambda p: p.as_posix())
if primary is not None:
@@ -176,9 +178,10 @@ def _calculate_classpath(roots: Sequence[pathlib.Path], *,
primary: pathlib.Path
jars = [primary, *(p for p in jars if p.resolve() != target)]
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(
+ raise TaskLaunchError(
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."
+ f"entries: {', '.join(map(os.fspath, bad))}. Rename or move them.",
+ retryable=False,
)
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}
@@ -391,6 +394,7 @@ class JavaCoordinator(SubprocessCoordinator):
JARs go on the classpath, so another JAR that sets the same
``Main-Class`` is rejected.
:raises ValueError: when the JAR cannot run.
+ :raises TaskLaunchError: when the bundle's JARs cannot be put on a
classpath.
"""
main_class, schema_version = _read_executable_jar(path)
if self.main_class and main_class != self.main_class:
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 e4a0e07df32..e3b1e3dadac 100644
--- a/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py
+++ b/task-sdk/tests/task_sdk/coordinators/java/test_coordinator.py
@@ -309,10 +309,12 @@ class TestClasspathLink:
with (
patch("airflow.sdk.coordinators.java.coordinator.os.getuid",
autospec=True, return_value=uid),
- pytest.raises(PermissionError, match="Cannot keep Java classpath
links"),
+ pytest.raises(TaskLaunchError, match="Cannot keep Java classpath
links") as ctx,
):
_get_classpath_link(root)
+ assert ctx.value.retryable is False
+
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"
@@ -324,10 +326,11 @@ class TestClasspathLink:
autospec=True,
return_value=os.fspath(tmp),
),
- pytest.raises(ValueError, match="Cannot keep Java classpath
links"),
+ pytest.raises(TaskLaunchError, match="Cannot keep Java classpath
links") as ctx,
):
_calculate_classpath([root])
+ assert ctx.value.retryable is False
assert list(tmp.iterdir()) == []
def test_separator_inside_the_bundle_raises_without_linking(self,
tmp_path, link_dir):
@@ -336,9 +339,10 @@ class TestClasspathLink:
nested.mkdir()
jar = _make_jar(nested / "app.jar")
- with pytest.raises(ValueError, match="Cannot put these JARs on a Java
classpath") as ctx:
+ with pytest.raises(TaskLaunchError, match="Cannot put these JARs on a
Java classpath") as ctx:
_calculate_classpath([root])
+ assert ctx.value.retryable is False
assert os.fspath(jar) in str(ctx.value)
assert not link_dir.exists()
