From 9a64c5ce357b16f85007791e963d40d7f76dda4d Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Tue, 7 Jul 2026 11:48:54 +0000 Subject: [PATCH 1/3] 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 --- .../commands/kubernetes_commands.py | 135 +++++++++++-- .../test_kubernetes_lang_sdk_commands.py | 186 ++++++++++++++++-- kubernetes-tests/lang_sdk/README.md | 11 ++ 3 files changed, 300 insertions(+), 32 deletions(-) diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py index 7b36aebe00b1a..31ba45bd230ab 100644 --- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py @@ -2497,6 +2497,7 @@ def deploy_cluster( 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 +2508,94 @@ def deploy_cluster( "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 + jar is ``export-ignore`` (ASF LEGAL-570) so ``git archive`` drops it; ``git show`` restores it. + """ + 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) + wrapper_jar = extracted / "java-sdk" / "gradle" / "wrapper" / "gradle-wrapper.jar" + wrapper_jar.parent.mkdir(parents=True, exist_ok=True) + wrapper_jar.write_bytes( + run_command( + ["git", "show", f"{sha}:java-sdk/gradle/wrapper/gradle-wrapper.jar"], + cwd=AIRFLOW_ROOT_PATH, + output=output, + capture_output=True, + check=True, + ).stdout + ) + (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 ``/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")) + 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 +2603,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 +2630,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, @@ -2570,7 +2649,9 @@ def _lang_sdk_build_go_bundle(staging: Path, output: Output | None, *, native: b 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 +2660,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 +2674,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 +2689,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 +2701,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( @@ -2921,9 +3011,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 +3026,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 651c0a17dc59f..4b6cf62573764 100644 --- a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py +++ b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py @@ -24,6 +24,7 @@ from airflow_breeze.commands.kubernetes_commands import ( _lang_sdk_build_go_bundle, _lang_sdk_build_java_jar, + _lang_sdk_fetch_upstream_sdk_sources, ) @@ -38,11 +39,21 @@ def go_example(tmp_path, monkeypatch): return go_dir +@pytest.fixture +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 +61,52 @@ def java_example(tmp_path, monkeypatch): return java_dir +@pytest.fixture +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 +124,126 @@ def test_native_uses_host_gradle_toolchain(self, mock_run, tmp_path, java_exampl "--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 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 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""), + ] + + 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""), + ] + + _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_jar_dropped_by_export_ignore(self, mock_run, tmp_path): + # export-ignore (ASF LEGAL-570) drops the jar from `git archive`; ./gradlew needs it 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-jar-bytes"), + ] + + _, java_sdk = _lang_sdk_fetch_upstream_sdk_sources(tmp_path, None) + + wrapper_jar = java_sdk / "gradle" / "wrapper" / "gradle-wrapper.jar" + assert wrapper_jar.read_bytes() == b"fake-jar-bytes" + show_cmd = mock_run.call_args_list[5].args[0] + assert show_cmd == ["git", "show", "deadbeef:java-sdk/gradle/wrapper/gradle-wrapper.jar"] class TestSetupLangSdkTestNativeSelection: @@ -107,28 +251,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 +301,9 @@ def fake_parallel(steps, output): 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 55e0d697935b5..f4840623c33f0 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 From 01d8cbd247c041c64f8e678f547ca426ec8e7417 Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Thu, 9 Jul 2026 10:06:55 +0000 Subject: [PATCH 2/3] 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. --- .../commands/kubernetes_commands.py | 31 ++++++++++++------- .../test_kubernetes_lang_sdk_commands.py | 30 ++++++++++++++---- 2 files changed, 43 insertions(+), 18 deletions(-) diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py index 31ba45bd230ab..843f3908c2dd9 100644 --- a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py +++ b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py @@ -2523,7 +2523,8 @@ def _lang_sdk_fetch_upstream_sdk_sources(staging: Path, output: Output | None) - 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 - jar is ``export-ignore`` (ASF LEGAL-570) so ``git archive`` drops it; ``git show`` restores it. + 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 @@ -2557,17 +2558,23 @@ def _lang_sdk_fetch_upstream_sdk_sources(staging: Path, output: Output | None) - check=True, ) run_command(["tar", "-xf", str(archive_path), "-C", str(extracted)], output=output, check=True) - wrapper_jar = extracted / "java-sdk" / "gradle" / "wrapper" / "gradle-wrapper.jar" - wrapper_jar.parent.mkdir(parents=True, exist_ok=True) - wrapper_jar.write_bytes( - run_command( - ["git", "show", f"{sha}:java-sdk/gradle/wrapper/gradle-wrapper.jar"], - cwd=AIRFLOW_ROOT_PATH, - output=output, - capture_output=True, - check=True, - ).stdout - ) + 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 + ) + restored.chmod(mode) (extracted / "task-sdk").symlink_to(AIRFLOW_ROOT_PATH / "task-sdk") return extracted / "go-sdk", extracted / "java-sdk" diff --git a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py index 4b6cf62573764..49c02e7f442e2 100644 --- a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py +++ b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py @@ -151,6 +151,8 @@ def test_prefers_upstream_remote_when_present(self, mock_run, tmp_path): 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 ] @@ -167,6 +169,8 @@ def test_falls_back_to_canonical_url_when_no_upstream_remote(self, mock_run, tmp 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 ] @@ -190,6 +194,8 @@ def test_returns_extracted_go_sdk_and_java_sdk_paths(self, mock_run, tmp_path): 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) @@ -218,6 +224,8 @@ def test_symlinks_real_task_sdk_alongside_the_extraction(self, mock_run, tmp_pat mock.Mock(), mock.Mock(), mock.Mock(stdout=b""), + mock.Mock(stdout=b""), + mock.Mock(stdout=b""), ] _lang_sdk_fetch_upstream_sdk_sources(staging, None) @@ -227,23 +235,33 @@ def test_symlinks_real_task_sdk_alongside_the_extraction(self, mock_run, tmp_pat assert task_sdk_link.resolve() == (repo_root / "task-sdk").resolve() @mock.patch.object(kubernetes_commands, "run_command") - def test_restores_gradle_wrapper_jar_dropped_by_export_ignore(self, mock_run, tmp_path): - # export-ignore (ASF LEGAL-570) drops the jar from `git archive`; ./gradlew needs it restored. + 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) - wrapper_jar = java_sdk / "gradle" / "wrapper" / "gradle-wrapper.jar" - assert wrapper_jar.read_bytes() == b"fake-jar-bytes" - show_cmd = mock_run.call_args_list[5].args[0] - assert show_cmd == ["git", "show", "deadbeef:java-sdk/gradle/wrapper/gradle-wrapper.jar"] + 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 TestSetupLangSdkTestNativeSelection: From 5b489ec12458d141fcbad571652466cfed1c3c0d Mon Sep 17 00:00:00 2001 From: LIU ZHE YOU Date: Mon, 13 Jul 2026 01:53:31 +0000 Subject: [PATCH 3/3] 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. --- .../commands/kubernetes_commands.py | 18 +++++++-- .../test_kubernetes_lang_sdk_commands.py | 40 +++++++++++++++++++ 2 files changed, 55 insertions(+), 3 deletions(-) diff --git a/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py b/dev/breeze/src/airflow_breeze/commands/kubernetes_commands.py index 843f3908c2dd9..ac5bbbfad045b 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 @@ check_if_image_exists, run_command, ) +from airflow_breeze.utils.shared_options import get_dry_run KUBERNETES_PYTEST_ARGS = [ "--strict-markers", @@ -2573,6 +2574,7 @@ def _lang_sdk_fetch_upstream_sdk_sources(staging: Path, output: Output | None) - capture_output=True, check=True, ).stdout + or b"" ) restored.chmod(mode) (extracted / "task-sdk").symlink_to(AIRFLOW_ROOT_PATH / "task-sdk") @@ -2600,7 +2602,10 @@ def _lang_sdk_build_go_bundle( 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")) - shutil.copytree(upstream_go_sdk, workspace / "go-sdk") + # 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) @@ -2653,7 +2658,8 @@ def _lang_sdk_build_go_bundle( 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( @@ -2753,6 +2759,8 @@ def _lang_sdk_build_java_jar( 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") @@ -2813,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 ( diff --git a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py index 49c02e7f442e2..ab5eb9946ea53 100644 --- a/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py +++ b/dev/breeze/tests/test_kubernetes_lang_sdk_commands.py @@ -25,7 +25,14 @@ _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 + + +@pytest.fixture +def dry_run(monkeypatch): + monkeypatch.setattr(shared_options._SharedOptions, "dry_run_value", True) @pytest.fixture @@ -264,6 +271,39 @@ def test_restores_gradle_wrapper_files_dropped_by_export_ignore(self, mock_run, ] +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: @pytest.mark.parametrize( ("env_value", "expected_native"),