This is an automated email from the ASF dual-hosted git repository.
jason810496 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 c22cc6d345f Pin lang-SDK k8s test's Go/Java SDKs to upstream main
(#69545)
c22cc6d345f is described below
commit c22cc6d345f3abe06d61cfc95fe851fbab8bd568
Author: Jason(Zhe-You) Liu <[email protected]>
AuthorDate: Wed Jul 15 11:43:46 2026 +0900
Pin lang-SDK k8s test's Go/Java SDKs to upstream main (#69545)
* Pin lang-SDK k8s test's Go/Java SDKs to upstream main
Release/backport branches often lack go-sdk/java-sdk entirely or carry a
stale, branch-cut-frozen copy, so backporting a core/task-sdk fix to such
a branch would otherwise run the lang-SDK coordinator test against a
missing or outdated SDK snapshot. The Go bundle and Java jar builds now
always fetch go-sdk/java-sdk from upstream main regardless of the checked
out branch, while airflow-core/task-sdk and the test's own harness
fixtures keep tracking whatever branch is checked out.
Fix lang-SDK Java build missing task-sdk sibling in native mode
java-sdk's sdk/build.gradle.kts reads a sibling ../task-sdk/.../schema.json
at build time. The upstream-main extraction only contains go-sdk/java-sdk,
so building from it directly (native/CI toolchain mode) failed looking for
a task-sdk directory that doesn't exist. Symlink the real, local task-sdk
alongside the extraction so the reference resolves, without pulling
task-sdk itself from upstream.
Restore lang-SDK k8s test's gradle wrapper jar dropped by export-ignore
java-sdk/.gitattributes marks gradle/wrapper/gradle-wrapper.jar
export-ignore
so ASF source-release tarballs stay free of compiled binaries (LEGAL-570).
git archive, used to pull go-sdk/java-sdk out of upstream main for this
test,
respects that attribute and silently drops the jar, leaving the extracted
java-sdk's ./gradlew unable to find itself and failing every build.
Shorten gradle-wrapper.jar restoration comments
* Restore gradle wrapper scripts dropped by export-ignore in lang-SDK k8s
test
Upstream main's java-sdk/.gitattributes export-ignores gradlew and
gradlew.bat (ASF LEGAL-570) in addition to the wrapper jar, so the
git-archive extraction of upstream sources no longer contains the
wrapper script the Java build invokes, failing CI with
FileNotFoundError: './gradlew'. Restore all three wrapper files from
the pinned commit instead of only the jar.
* Fix breeze k8s setup-lang-sdk-test crash with --dry-run
In dry-run mode run_command skips execution and returns stdout="" (a
str), so every filesystem step consuming a command's output crashed:
write_bytes on the gradle wrapper restore raised TypeError, the Go
bundle copytree/copy raised FileNotFoundError, the Java jar glob hit
sys.exit(1), and the artifact upload raised StopIteration. Guard those
steps so --dry-run prints the full provisioning command sequence and
completes cleanly.
---
.../airflow_breeze/commands/kubernetes_commands.py | 158 +++++++++++--
.../tests/test_kubernetes_lang_sdk_commands.py | 244 +++++++++++++++++++--
kubernetes-tests/lang_sdk/README.md | 11 +
3 files changed, 379 insertions(+), 34 deletions(-)
diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
index 7b36aebe00b..ac5bbbfad04 100644
--- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
+++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py
@@ -100,6 +100,7 @@ from airflow_breeze.utils.run_utils import (
check_if_image_exists,
run_command,
)
+from airflow_breeze.utils.shared_options import get_dry_run
KUBERNETES_PYTEST_ARGS = [
"--strict-markers",
@@ -2497,6 +2498,7 @@ LANG_SDK_JAVA_EXAMPLE_PATH = LANG_SDK_PATH /
"java_example"
LANG_SDK_GO_BUILDER_IMAGE = os.environ.get("GO_BUILDER_IMAGE",
"golang:1.25-alpine")
LANG_SDK_JAVA_BUILDER_IMAGE = "eclipse-temurin:17-jdk"
LANG_SDK_MAVEN_CACHE_PATH = AIRFLOW_ROOT_PATH / "files" / "m2"
+LANG_SDK_GRADLE_CACHE_PATH = AIRFLOW_ROOT_PATH / "files" / "gradle"
# The Java queue needs a JRE the JavaCoordinator can exec; the Go queue runs on
# the plain prod image. Building the Java worker image as a separate tag (prod
+
# JRE, see Dockerfile.java) lets each coordinator route its queue to a distinct
@@ -2507,19 +2509,105 @@ LANG_SDK_AWS_CONN_URI = (
"aws://test:test@/?region_name=us-east-1&"
"endpoint_url=http%3A%2F%2Flocalstack.airflow.svc.cluster.local%3A4566"
)
+# The Go/Java SDKs are always built from upstream main so branches with stale
or missing
+# go-sdk/java-sdk copies still test current SDK sources. See
kubernetes-tests/lang_sdk/README.md.
+LANG_SDK_UPSTREAM_GIT_URL = "https://github.com/apache/airflow.git";
+LANG_SDK_UPSTREAM_REF = "main"
-def _lang_sdk_build_go_bundle(staging: Path, output: Output | None, *, native:
bool = False) -> None:
+def _lang_sdk_fetch_upstream_sdk_sources(staging: Path, output: Output | None)
-> tuple[Path, Path]:
+ """Extract go-sdk/ and java-sdk/ from upstream main into a throwaway
staging dir.
+
+ Prefers the ``upstream`` remote when configured, falling back to the
canonical GitHub URL
+ (CI has no ``upstream`` and ``origin`` may be a fork). Shallow fetch +
``git archive`` never
+ touch the working tree or index.
+
+ The real, local task-sdk is symlinked alongside the extraction because
java-sdk's
+ ``sdk/build.gradle.kts`` reads a sibling ``../task-sdk/.../schema.json``.
The gradle wrapper
+ scripts and jar are ``export-ignore`` (ASF LEGAL-570) so ``git archive``
drops them;
+ ``git show`` restores them, re-marking ``gradlew`` executable.
+ """
+ remotes = run_command(
+ ["git", "remote"], cwd=AIRFLOW_ROOT_PATH, output=output,
capture_output=True, text=True, check=True
+ ).stdout.split()
+ fetch_source = "upstream" if "upstream" in remotes else
LANG_SDK_UPSTREAM_GIT_URL
+ get_console(output=output).print(
+ f"[info]Fetching {LANG_SDK_UPSTREAM_REF} from {fetch_source} for the
lang-SDK Go/Java sources"
+ )
+ run_command(
+ ["git", "fetch", "--depth=1", fetch_source, LANG_SDK_UPSTREAM_REF],
+ cwd=AIRFLOW_ROOT_PATH,
+ output=output,
+ check=True,
+ )
+ sha = run_command(
+ ["git", "rev-parse", "FETCH_HEAD"],
+ cwd=AIRFLOW_ROOT_PATH,
+ output=output,
+ capture_output=True,
+ text=True,
+ check=True,
+ ).stdout.strip()
+ get_console(output=output).print(f"[info]lang-SDK Go/Java sources pinned
to upstream main @ {sha}")
+ extracted = staging / "upstream_lang_sdk_sources"
+ extracted.mkdir(parents=True, exist_ok=True)
+ archive_path = staging / "upstream_lang_sdk_sources.tar"
+ run_command(
+ ["git", "archive", "--format=tar", f"--output={archive_path}", sha,
"--", "go-sdk", "java-sdk"],
+ cwd=AIRFLOW_ROOT_PATH,
+ output=output,
+ check=True,
+ )
+ run_command(["tar", "-xf", str(archive_path), "-C", str(extracted)],
output=output, check=True)
+ for rel_path, mode in (
+ ("java-sdk/gradlew", 0o755),
+ ("java-sdk/gradlew.bat", 0o644),
+ ("java-sdk/gradle/wrapper/gradle-wrapper.jar", 0o644),
+ ):
+ restored = extracted / rel_path
+ restored.parent.mkdir(parents=True, exist_ok=True)
+ restored.write_bytes(
+ run_command(
+ ["git", "show", f"{sha}:{rel_path}"],
+ cwd=AIRFLOW_ROOT_PATH,
+ output=output,
+ capture_output=True,
+ check=True,
+ ).stdout
+ or b""
+ )
+ restored.chmod(mode)
+ (extracted / "task-sdk").symlink_to(AIRFLOW_ROOT_PATH / "task-sdk")
+ return extracted / "go-sdk", extracted / "java-sdk"
+
+
+def _lang_sdk_build_go_bundle(
+ staging: Path, upstream_go_sdk: Path, output: Output | None, *, native:
bool = False
+) -> None:
"""Build the Go bundle into ``staging/go-artifacts`` and copy the result
into the staging dir.
By default the build runs in an ephemeral Go toolchain container so the
host needs no Go install,
writing its caches into a gitignored dir under the repo. In ``native``
mode (used in CI, where the
host already has a cached Go toolchain via ``actions/setup-go``) it
invokes the host ``go`` directly,
skipping the container image pull and reusing the runner's module/build
cache.
+
+ go_example's go.mod ``replace``s go-sdk by relative path, so the build
runs in a scratch
+ workspace mirroring the repo layout with ``upstream_go_sdk`` at
``<workspace>/go-sdk``,
+ letting the unmodified directive resolve against the upstream copy.
"""
go_dir = staging / "go-artifacts"
go_dir.mkdir(parents=True, exist_ok=True)
- output_bin = LANG_SDK_GO_EXAMPLE_PATH / "bin" / LANG_SDK_GO_BUNDLE_NAME
+ example_rel = LANG_SDK_GO_EXAMPLE_PATH.relative_to(AIRFLOW_ROOT_PATH)
+ workspace = staging / "go_workspace"
+ example_path = workspace / example_rel
+ example_path.parent.mkdir(parents=True, exist_ok=True)
+ shutil.copytree(LANG_SDK_GO_EXAMPLE_PATH, example_path,
ignore=shutil.ignore_patterns(".home"))
+ # In dry-run the fetch/extract commands are skipped, so the upstream copy
and build outputs
+ # never materialize -- skip the filesystem work that depends on them.
+ if not get_dry_run():
+ shutil.copytree(upstream_go_sdk, workspace / "go-sdk")
+ output_bin = example_path / "bin" / LANG_SDK_GO_BUNDLE_NAME
+ output_bin.parent.mkdir(parents=True, exist_ok=True)
# CGO_ENABLED=0 yields a fully static binary that runs on the stock
worker. The package built is
# the current dir (".") because go_example is its own module.
@@ -2527,17 +2615,18 @@ def _lang_sdk_build_go_bundle(staging: Path, output:
Output | None, *, native: b
get_console(output=output).print("[info]Building Go bundle with the
host Go toolchain")
run_command(
["go", "tool", "airflow-go-pack", "--output", str(output_bin),
"."],
- cwd=LANG_SDK_GO_EXAMPLE_PATH,
+ cwd=example_path,
env={**os.environ, "CGO_ENABLED": "0"},
output=output,
check=True,
)
else:
uid_gid = f"{os.getuid()}:{os.getgid()}"
- go_example_ctr =
f"/repo/{LANG_SDK_GO_EXAMPLE_PATH.relative_to(AIRFLOW_ROOT_PATH).as_posix()}"
+ go_example_ctr = f"/repo/{example_rel.as_posix()}"
# USER/HOME must be set because the SDK calls user.Current() at init;
with cgo disabled Go's
- # pure-Go resolver reads those env vars and panics if either is empty.
HOME points at a
- # writable, gitignored dir under go_example so the caches persist
between runs on a dev host.
+ # pure-Go resolver reads those env vars and panics if either is empty.
HOME is mounted from
+ # the real go_example's gitignored cache dir so the caches persist
across scratch workspaces.
+ (LANG_SDK_GO_EXAMPLE_PATH / ".home").mkdir(parents=True, exist_ok=True)
get_console(output=output).print(f"[info]Building Go bundle in
{LANG_SDK_GO_BUILDER_IMAGE}")
run_command(
[
@@ -2553,7 +2642,9 @@ def _lang_sdk_build_go_bundle(staging: Path, output:
Output | None, *, native: b
"-e",
"CGO_ENABLED=0",
"-v",
- f"{AIRFLOW_ROOT_PATH}:/repo",
+ f"{workspace}:/repo",
+ "-v",
+ f"{LANG_SDK_GO_EXAMPLE_PATH / '.home'}:{go_example_ctr}/.home",
"-w",
go_example_ctr,
LANG_SDK_GO_BUILDER_IMAGE,
@@ -2567,10 +2658,13 @@ def _lang_sdk_build_go_bundle(staging: Path, output:
Output | None, *, native: b
output=output,
check=True,
)
- shutil.copy(output_bin, go_dir / LANG_SDK_GO_BUNDLE_NAME)
+ if not get_dry_run():
+ shutil.copy(output_bin, go_dir / LANG_SDK_GO_BUNDLE_NAME)
-def _lang_sdk_build_java_jar(staging: Path, output: Output | None, *, native:
bool = False) -> None:
+def _lang_sdk_build_java_jar(
+ staging: Path, upstream_java_sdk: Path, output: Output | None, *, native:
bool = False
+) -> None:
"""Publish the Java SDK to mavenLocal then build the java_example jar into
``staging/java-artifacts``.
By default the build runs in an ephemeral JDK container so the host needs
no JDK, persisting the
@@ -2579,10 +2673,13 @@ def _lang_sdk_build_java_jar(staging: Path, output:
Output | None, *, native: bo
host ``./gradlew`` directly, skipping the container image pull and reusing
the runner's ``~/.gradle``
cache. ``java_example`` resolves the SDK from ``mavenLocal()``, so the SDK
is published first, then
the bundle is built with java-sdk's gradle wrapper pointed at the example
project (``-p``).
+
+ Both gradle invocations run against ``upstream_java_sdk`` rather than the
local ``java-sdk/``;
+ only ``-p`` stays pointed at the local ``java_example``, which is
test-harness code that keeps
+ tracking the checked-out branch.
"""
java_dir = staging / "java-artifacts"
java_dir.mkdir(parents=True, exist_ok=True)
- java_sdk_path = AIRFLOW_ROOT_PATH / "java-sdk"
if native:
get_console(output=output).print(
@@ -2590,14 +2687,14 @@ def _lang_sdk_build_java_jar(staging: Path, output:
Output | None, *, native: bo
)
run_command(
["./gradlew", "publishToMavenLocal", "-PskipSigning=true",
"--no-daemon", "--console=plain"],
- cwd=java_sdk_path,
+ cwd=upstream_java_sdk,
output=output,
check=True,
)
get_console(output=output).print("[info]Building Java jar with the
host Gradle toolchain")
run_command(
["./gradlew", "-p", str(LANG_SDK_JAVA_EXAMPLE_PATH), "bundle",
"--no-daemon", "--console=plain"],
- cwd=java_sdk_path,
+ cwd=upstream_java_sdk,
output=output,
check=True,
)
@@ -2605,9 +2702,11 @@ def _lang_sdk_build_java_jar(staging: Path, output:
Output | None, *, native: bo
uid_gid = f"{os.getuid()}:{os.getgid()}"
java_example_ctr =
f"/repo/{LANG_SDK_JAVA_EXAMPLE_PATH.relative_to(AIRFLOW_ROOT_PATH).as_posix()}"
# --user keeps build outputs owned by the host user; HOME is set
explicitly because that UID has
- # no /etc/passwd entry; GRADLE_USER_HOME and the mounted ~/.m2 persist
the Gradle distribution and
- # dependency caches between runs on a dev host.
+ # no /etc/passwd entry. GRADLE_USER_HOME and the mounted ~/.m2 persist
the Gradle distribution
+ # and dependency caches between runs -- outside /repo/java-sdk, which
is remounted to the
+ # (per-run) upstream copy below.
LANG_SDK_MAVEN_CACHE_PATH.mkdir(parents=True, exist_ok=True)
+ LANG_SDK_GRADLE_CACHE_PATH.mkdir(parents=True, exist_ok=True)
java_docker_prefix = [
"docker",
"run",
@@ -2615,13 +2714,17 @@ def _lang_sdk_build_java_jar(staging: Path, output:
Output | None, *, native: bo
"--user",
uid_gid,
"-e",
- "GRADLE_USER_HOME=/repo/java-sdk/.gradle",
+ "GRADLE_USER_HOME=/workspace-home/.gradle",
"-e",
"HOME=/workspace-home",
"-v",
f"{LANG_SDK_MAVEN_CACHE_PATH}:/workspace-home/.m2",
"-v",
+ f"{LANG_SDK_GRADLE_CACHE_PATH}:/workspace-home/.gradle",
+ "-v",
f"{AIRFLOW_ROOT_PATH}:/repo",
+ "-v",
+ f"{upstream_java_sdk}:/repo/java-sdk",
]
get_console(output=output).print("[info]Publishing Java SDK artifacts
to local Maven repository")
run_command(
@@ -2656,6 +2759,8 @@ def _lang_sdk_build_java_jar(staging: Path, output:
Output | None, *, native: bo
output=output,
check=True,
)
+ if get_dry_run():
+ return
jars = list((LANG_SDK_JAVA_EXAMPLE_PATH / "build" /
"bundle").glob("*.jar"))
if not jars:
get_console(output=output).print("[error]No jar produced by the Java
bundle build")
@@ -2716,7 +2821,11 @@ def _lang_sdk_upload_artifacts(
).stdout.strip()
go_bundle = staging / "go-artifacts" / "lang_sdk_combined"
- java_jar = next((staging / "java-artifacts").glob("*.jar"))
+ if get_dry_run():
+ # The dry-run build steps produce no jar; use the placeholder name so
the commands still print.
+ java_jar = staging / "java-artifacts" / "app.jar"
+ else:
+ java_jar = next((staging / "java-artifacts").glob("*.jar"))
stub_dag = LANG_SDK_PATH / "dags" / "lang_sdk_combined.py"
for src, dest in (
@@ -2921,9 +3030,9 @@ def _setup_lang_sdk_test(
) -> None:
"""Provision the lang-SDK coordinator env on an already-deployed
KubernetesExecutor cluster.
- Builds the Go/Java artifacts, the Java worker image and deploys localstack
in parallel, then
- serially uploads the artifacts, applies the config + secret, and
helm-upgrades Airflow with the
- lang-SDK values.
+ Fetches go-sdk/java-sdk from upstream main, then builds the Go/Java
artifacts, the Java worker
+ image and deploys localstack in parallel, then serially uploads the
artifacts, applies the
+ config + secret, and helm-upgrades Airflow with the lang-SDK values.
"""
go_image = go_image or
f"{BuildProdParams(python=python).airflow_image_kubernetes}:latest"
build_java_image = java_image is None
@@ -2936,9 +3045,16 @@ def _setup_lang_sdk_test(
native = os.environ.get("LANG_SDK_NATIVE_TOOLCHAIN", "").lower() == "true"
with tempfile.TemporaryDirectory(prefix="lang_sdk_artifacts_") as tmp:
staging = Path(tmp)
+ upstream_go_sdk, upstream_java_sdk =
_lang_sdk_fetch_upstream_sdk_sources(staging, output)
steps: list[tuple[str, Callable[[Output | None], Any]]] = [
- ("Build Go bundle", lambda o: _lang_sdk_build_go_bundle(staging,
o, native=native)),
- ("Build Java jar", lambda o: _lang_sdk_build_java_jar(staging, o,
native=native)),
+ (
+ "Build Go bundle",
+ lambda o: _lang_sdk_build_go_bundle(staging, upstream_go_sdk,
o, native=native),
+ ),
+ (
+ "Build Java jar",
+ lambda o: _lang_sdk_build_java_jar(staging, upstream_java_sdk,
o, native=native),
+ ),
("Deploy localstack", lambda o:
_lang_sdk_deploy_localstack(python, kubernetes_version, o)),
]
if build_java_image:
diff --git a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
index 651c0a17dc5..ab5eb9946ea 100644
--- a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
+++ b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py
@@ -24,7 +24,15 @@ from airflow_breeze.commands import kubernetes_commands
from airflow_breeze.commands.kubernetes_commands import (
_lang_sdk_build_go_bundle,
_lang_sdk_build_java_jar,
+ _lang_sdk_fetch_upstream_sdk_sources,
+ _lang_sdk_upload_artifacts,
)
+from airflow_breeze.utils import shared_options
+
+
[email protected]
+def dry_run(monkeypatch):
+ monkeypatch.setattr(shared_options._SharedOptions, "dry_run_value", True)
@pytest.fixture
@@ -38,11 +46,21 @@ def go_example(tmp_path, monkeypatch):
return go_dir
[email protected]
+def upstream_go_sdk(tmp_path):
+ """A fake extracted upstream-main go-sdk copy, distinguishable from
anything under go_example."""
+ sdk_dir = tmp_path / "upstream_go_sdk"
+ sdk_dir.mkdir()
+ (sdk_dir / "marker.go").write_text("upstream")
+ return sdk_dir
+
+
@pytest.fixture
def java_example(tmp_path, monkeypatch):
"""Point the java_example dir at a tmp path (under a tmp repo root) with a
pre-built bundle jar."""
monkeypatch.setattr(kubernetes_commands, "AIRFLOW_ROOT_PATH", tmp_path)
monkeypatch.setattr(kubernetes_commands, "LANG_SDK_MAVEN_CACHE_PATH",
tmp_path / "m2")
+ monkeypatch.setattr(kubernetes_commands, "LANG_SDK_GRADLE_CACHE_PATH",
tmp_path / "gradle")
java_dir = tmp_path / "java_example"
(java_dir / "build" / "bundle").mkdir(parents=True)
(java_dir / "build" / "bundle" / "app.jar").write_text("jar")
@@ -50,31 +68,52 @@ def java_example(tmp_path, monkeypatch):
return java_dir
[email protected]
+def upstream_java_sdk(tmp_path):
+ """A fake extracted upstream-main java-sdk copy, distinguishable from the
real java-sdk/."""
+ sdk_dir = tmp_path / "upstream_java_sdk"
+ sdk_dir.mkdir()
+ (sdk_dir / "marker.gradle").write_text("upstream")
+ return sdk_dir
+
+
class TestLangSdkBuildGoBundle:
@mock.patch.object(kubernetes_commands, "run_command")
- def test_native_uses_host_go_toolchain(self, mock_run, tmp_path,
go_example):
- _lang_sdk_build_go_bundle(tmp_path, None, native=True)
+ def test_native_uses_host_go_toolchain(self, mock_run, tmp_path,
go_example, upstream_go_sdk):
+ _lang_sdk_build_go_bundle(tmp_path, upstream_go_sdk, None, native=True)
cmd = mock_run.call_args.args[0]
assert cmd[:3] == ["go", "tool", "airflow-go-pack"]
assert "docker" not in cmd
- assert mock_run.call_args.kwargs["cwd"] == go_example
+ workspace_example = mock_run.call_args.kwargs["cwd"]
+ # The build runs against a scratch workspace copy, not the real
go_example dir.
+ assert workspace_example != go_example
+ assert workspace_example.name == go_example.name
assert mock_run.call_args.kwargs["env"]["CGO_ENABLED"] == "0"
+ # The workspace mirrors the repo layout, with go-sdk swapped for the
upstream copy.
+ assert (workspace_example.parent / "go-sdk" / "marker.go").read_text()
== "upstream"
assert (tmp_path / "go-artifacts" /
kubernetes_commands.LANG_SDK_GO_BUNDLE_NAME).exists()
@mock.patch.object(kubernetes_commands, "run_command")
- def test_container_mode_runs_in_docker(self, mock_run, tmp_path,
go_example):
- _lang_sdk_build_go_bundle(tmp_path, None, native=False)
+ def test_container_mode_runs_in_docker(self, mock_run, tmp_path,
go_example, upstream_go_sdk):
+ _lang_sdk_build_go_bundle(tmp_path, upstream_go_sdk, None,
native=False)
cmd = mock_run.call_args.args[0]
assert cmd[0] == "docker"
assert kubernetes_commands.LANG_SDK_GO_BUILDER_IMAGE in cmd
+ # The workspace, not AIRFLOW_ROOT_PATH, is mounted at /repo; the
persistent HOME cache
+ # dir comes from the real go_example.
+ mounts = [cmd[i + 1] for i, arg in enumerate(cmd) if arg == "-v"]
+ repo_mount = next(m for m in mounts if m.endswith(":/repo"))
+ assert repo_mount.split(":")[0] != str(go_example.parent)
+ home_mount = next(m for m in mounts if m.endswith("/.home"))
+ assert home_mount.startswith(str(go_example / ".home"))
class TestLangSdkBuildJavaJar:
@mock.patch.object(kubernetes_commands, "run_command")
- def test_native_uses_host_gradle_toolchain(self, mock_run, tmp_path,
java_example):
- _lang_sdk_build_java_jar(tmp_path, None, native=True)
+ def test_native_uses_host_gradle_toolchain(self, mock_run, tmp_path,
java_example, upstream_java_sdk):
+ _lang_sdk_build_java_jar(tmp_path, upstream_java_sdk, None,
native=True)
publish_cmd, bundle_cmd = (call.args[0] for call in
mock_run.call_args_list)
assert publish_cmd == [
@@ -92,14 +131,177 @@ class TestLangSdkBuildJavaJar:
"--no-daemon",
"--console=plain",
]
+ # gradlew runs from the upstream copy; only -p stays pointed at the
local java_example.
+ assert all(call.kwargs["cwd"] == upstream_java_sdk for call in
mock_run.call_args_list)
assert all("docker" not in call.args[0] for call in
mock_run.call_args_list)
assert (tmp_path / "java-artifacts" / "app.jar").exists()
@mock.patch.object(kubernetes_commands, "run_command")
- def test_container_mode_runs_in_docker(self, mock_run, tmp_path,
java_example):
- _lang_sdk_build_java_jar(tmp_path, None, native=False)
+ def test_container_mode_runs_in_docker(self, mock_run, tmp_path,
java_example, upstream_java_sdk):
+ _lang_sdk_build_java_jar(tmp_path, upstream_java_sdk, None,
native=False)
assert all(call.args[0][0] == "docker" for call in
mock_run.call_args_list)
+ cmd = mock_run.call_args_list[0].args[0]
+ mounts = [cmd[i + 1] for i, arg in enumerate(cmd) if arg == "-v"]
+ # java-sdk is remounted to the upstream copy; GRADLE_USER_HOME lives
outside it.
+ assert f"{upstream_java_sdk}:/repo/java-sdk" in mounts
+ assert "GRADLE_USER_HOME=/workspace-home/.gradle" in cmd
+ assert any(mount.endswith(":/workspace-home/.gradle") for mount in
mounts)
+
+
+class TestLangSdkFetchUpstreamSdkSources:
+ @mock.patch.object(kubernetes_commands, "run_command")
+ def test_prefers_upstream_remote_when_present(self, mock_run, tmp_path):
+ mock_run.side_effect = [
+ mock.Mock(stdout="origin\nupstream\n"),
+ mock.Mock(), # git fetch
+ mock.Mock(stdout="deadbeef\n"), # git rev-parse
+ mock.Mock(), # git archive
+ mock.Mock(), # tar -xf
+ mock.Mock(stdout=b""), # git show gradlew
+ mock.Mock(stdout=b""), # git show gradlew.bat
+ mock.Mock(stdout=b""), # git show gradle-wrapper.jar
+ ]
+
+ _lang_sdk_fetch_upstream_sdk_sources(tmp_path, None)
+
+ fetch_cmd = mock_run.call_args_list[1].args[0]
+ assert fetch_cmd == ["git", "fetch", "--depth=1", "upstream", "main"]
+
+ @mock.patch.object(kubernetes_commands, "run_command")
+ def test_falls_back_to_canonical_url_when_no_upstream_remote(self,
mock_run, tmp_path):
+ mock_run.side_effect = [
+ mock.Mock(stdout="origin\n"),
+ mock.Mock(), # git fetch
+ mock.Mock(stdout="deadbeef\n"), # git rev-parse
+ mock.Mock(), # git archive
+ mock.Mock(), # tar -xf
+ mock.Mock(stdout=b""), # git show gradlew
+ mock.Mock(stdout=b""), # git show gradlew.bat
+ mock.Mock(stdout=b""), # git show gradle-wrapper.jar
+ ]
+
+ _lang_sdk_fetch_upstream_sdk_sources(tmp_path, None)
+
+ fetch_cmd = mock_run.call_args_list[1].args[0]
+ assert fetch_cmd == [
+ "git",
+ "fetch",
+ "--depth=1",
+ "https://github.com/apache/airflow.git";,
+ "main",
+ ]
+
+ @mock.patch.object(kubernetes_commands, "run_command")
+ def test_returns_extracted_go_sdk_and_java_sdk_paths(self, mock_run,
tmp_path):
+ mock_run.side_effect = [
+ mock.Mock(stdout="upstream\n"),
+ mock.Mock(),
+ mock.Mock(stdout="deadbeef\n"),
+ mock.Mock(),
+ mock.Mock(),
+ mock.Mock(stdout=b""),
+ mock.Mock(stdout=b""),
+ mock.Mock(stdout=b""),
+ ]
+
+ go_sdk, java_sdk = _lang_sdk_fetch_upstream_sdk_sources(tmp_path, None)
+
+ extracted = tmp_path / "upstream_lang_sdk_sources"
+ assert go_sdk == extracted / "go-sdk"
+ assert java_sdk == extracted / "java-sdk"
+ archive_cmd = mock_run.call_args_list[3].args[0]
+ assert archive_cmd[:2] == ["git", "archive"]
+ assert archive_cmd[-2:] == ["go-sdk", "java-sdk"]
+ assert "deadbeef" in archive_cmd
+
+ @mock.patch.object(kubernetes_commands, "run_command")
+ def test_symlinks_real_task_sdk_alongside_the_extraction(self, mock_run,
tmp_path, monkeypatch):
+ # java-sdk's build reads a sibling ../task-sdk/.../schema.json;
without the symlink a
+ # native-mode build from the extraction fails.
+ repo_root = tmp_path / "repo"
+ (repo_root / "task-sdk").mkdir(parents=True)
+ monkeypatch.setattr(kubernetes_commands, "AIRFLOW_ROOT_PATH",
repo_root)
+ staging = tmp_path / "staging"
+ staging.mkdir()
+ mock_run.side_effect = [
+ mock.Mock(stdout="upstream\n"),
+ mock.Mock(),
+ mock.Mock(stdout="deadbeef\n"),
+ mock.Mock(),
+ mock.Mock(),
+ mock.Mock(stdout=b""),
+ mock.Mock(stdout=b""),
+ mock.Mock(stdout=b""),
+ ]
+
+ _lang_sdk_fetch_upstream_sdk_sources(staging, None)
+
+ task_sdk_link = staging / "upstream_lang_sdk_sources" / "task-sdk"
+ assert task_sdk_link.is_symlink()
+ assert task_sdk_link.resolve() == (repo_root / "task-sdk").resolve()
+
+ @mock.patch.object(kubernetes_commands, "run_command")
+ def test_restores_gradle_wrapper_files_dropped_by_export_ignore(self,
mock_run, tmp_path):
+ # export-ignore (ASF LEGAL-570) drops gradlew, gradlew.bat, and the
wrapper jar from
+ # `git archive`; the build invokes ./gradlew from the extraction, so
all must be restored.
+ mock_run.side_effect = [
+ mock.Mock(stdout="upstream\n"),
+ mock.Mock(),
+ mock.Mock(stdout="deadbeef\n"),
+ mock.Mock(),
+ mock.Mock(),
+ mock.Mock(stdout=b"fake-gradlew"),
+ mock.Mock(stdout=b"fake-gradlew-bat"),
+ mock.Mock(stdout=b"fake-jar-bytes"),
+ ]
+
+ _, java_sdk = _lang_sdk_fetch_upstream_sdk_sources(tmp_path, None)
+
+ gradlew = java_sdk / "gradlew"
+ assert gradlew.read_bytes() == b"fake-gradlew"
+ assert gradlew.stat().st_mode & 0o111, "gradlew must be executable"
+ assert (java_sdk / "gradlew.bat").read_bytes() == b"fake-gradlew-bat"
+ assert (java_sdk / "gradle" / "wrapper" /
"gradle-wrapper.jar").read_bytes() == b"fake-jar-bytes"
+ show_cmds = [call.args[0] for call in mock_run.call_args_list[5:8]]
+ assert show_cmds == [
+ ["git", "show", "deadbeef:java-sdk/gradlew"],
+ ["git", "show", "deadbeef:java-sdk/gradlew.bat"],
+ ["git", "show",
"deadbeef:java-sdk/gradle/wrapper/gradle-wrapper.jar"],
+ ]
+
+
+class TestLangSdkDryRun:
+ """Dry-run skips the commands, so the filesystem work depending on their
outputs must not run."""
+
+ def test_fetch_upstream_sdk_sources_does_not_crash_on_str_stdout(self,
dry_run, tmp_path, monkeypatch):
+ monkeypatch.setattr(kubernetes_commands, "AIRFLOW_ROOT_PATH", tmp_path
/ "repo")
+
+ _, java_sdk = _lang_sdk_fetch_upstream_sdk_sources(tmp_path, None)
+
+ # Regression: dry-run run_command returns stdout="" (str) and
write_bytes("") raised TypeError.
+ assert (java_sdk / "gradlew").read_bytes() == b""
+
+ def test_build_go_bundle_skips_copies_of_never_built_artifacts(self,
dry_run, tmp_path, go_example):
+ _lang_sdk_build_go_bundle(tmp_path, tmp_path /
"missing_upstream_go_sdk", None, native=True)
+
+ assert not (tmp_path / "go-artifacts" /
kubernetes_commands.LANG_SDK_GO_BUNDLE_NAME).exists()
+
+ def test_build_java_jar_skips_jar_copy(self, dry_run, tmp_path,
java_example, upstream_java_sdk):
+ (java_example / "build" / "bundle" / "app.jar").unlink()
+
+ _lang_sdk_build_java_jar(tmp_path, upstream_java_sdk, None,
native=True)
+
+ assert not (tmp_path / "java-artifacts" / "app.jar").exists()
+
+ @mock.patch.object(kubernetes_commands, "run_command_with_k8s_env")
+ def test_upload_artifacts_uses_placeholder_for_never_built_jar(self,
mock_run, dry_run, tmp_path):
+ mock_run.return_value = mock.Mock(stdout="")
+
+ _lang_sdk_upload_artifacts(tmp_path, "3.10", "v1.35.0", None)
+
+ cp_sources = [call.args[0][2] for call in mock_run.call_args_list if
call.args[0][1] == "cp"]
+ assert str(tmp_path / "java-artifacts" / "app.jar") in cp_sources
class TestSetupLangSdkTestNativeSelection:
@@ -107,28 +309,39 @@ class TestSetupLangSdkTestNativeSelection:
("env_value", "expected_native"),
[("true", True), ("True", True), ("false", False), ("", False), (None,
False)],
)
- def test_native_flag_is_read_from_env(self, monkeypatch, env_value,
expected_native):
+ def test_native_flag_is_read_from_env(self, monkeypatch, tmp_path,
env_value, expected_native):
if env_value is None:
monkeypatch.delenv("LANG_SDK_NATIVE_TOOLCHAIN", raising=False)
else:
monkeypatch.setenv("LANG_SDK_NATIVE_TOOLCHAIN", env_value)
captured: dict[str, bool] = {}
+ fake_go_sdk = tmp_path / "upstream_go_sdk"
+ fake_java_sdk = tmp_path / "upstream_java_sdk"
def fake_parallel(steps, output):
for _title, thunk in steps:
thunk(None)
monkeypatch.setattr(kubernetes_commands, "_run_lang_sdk_parallel",
fake_parallel)
+ monkeypatch.setattr(
+ kubernetes_commands,
+ "_lang_sdk_fetch_upstream_sdk_sources",
+ lambda staging, output: (fake_go_sdk, fake_java_sdk),
+ )
monkeypatch.setattr(
kubernetes_commands,
"_lang_sdk_build_go_bundle",
- lambda staging, output, *, native: captured.update(go=native),
+ lambda staging, upstream_go_sdk, output, *, native:
captured.update(
+ go=native, go_sdk=upstream_go_sdk
+ ),
)
monkeypatch.setattr(
kubernetes_commands,
"_lang_sdk_build_java_jar",
- lambda staging, output, *, native: captured.update(java=native),
+ lambda staging, upstream_java_sdk, output, *, native:
captured.update(
+ java=native, java_sdk=upstream_java_sdk
+ ),
)
for name in (
"_lang_sdk_deploy_localstack",
@@ -146,4 +359,9 @@ class TestSetupLangSdkTestNativeSelection:
kubernetes_commands._setup_lang_sdk_test(python="3.10",
kubernetes_version="v1.35.0")
- assert captured == {"go": expected_native, "java": expected_native}
+ assert captured == {
+ "go": expected_native,
+ "java": expected_native,
+ "go_sdk": fake_go_sdk,
+ "java_sdk": fake_java_sdk,
+ }
diff --git a/kubernetes-tests/lang_sdk/README.md
b/kubernetes-tests/lang_sdk/README.md
index 55e0d697935..f4840623c33 100644
--- a/kubernetes-tests/lang_sdk/README.md
+++ b/kubernetes-tests/lang_sdk/README.md
@@ -63,6 +63,17 @@ coordinator scans.
The Go binary, Java jar, and stub Dag share one object store (localstack) but
live in
**separate buckets** (`go-artifacts`, `java-artifacts`, `dags`).
+## SDK sources always come from upstream main
+
+`go-sdk/` and `java-sdk/` are young, fast-moving directories that a
release/backport branch may lack
+entirely or carry a stale, branch-cut-frozen copy of. So regardless of which
branch/ref this repo is
+checked out at, `breeze k8s setup-lang-sdk-test` (and `run-complete-tests
--lang-sdk-test`) always
+builds the Go bundle and Java jar from upstream `main`'s `go-sdk`/`java-sdk` —
fetched fresh via
+`_lang_sdk_fetch_upstream_sdk_sources()` in `kubernetes_commands.py`, never
from whatever's on disk.
+Everything else — `airflow-core/`, `task-sdk/`, the deployed Airflow image,
and this directory's own
+`go_example`/`java_example` harness fixtures — still comes from the
checked-out branch as before, so a
+backport of a core/task-sdk fix to a release-test branch keeps testing against
current SDK code.
+
## Running it
The artifacts, localstack, config, and Helm release are provisioned by a
single breeze
