diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 4dfa0aa7fc214..50f3bdadcb015 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -280,6 +280,21 @@ repos: (?x) ^java-sdk/gradle\.properties$| ^java-sdk/sdk/schema/schema\.json$ + - id: update-java-sdk-readme-matrix + name: Update the Java SDK compatibility matrix in java-sdk/README.md and Dokka module doc + entry: ./scripts/ci/prek/update_java_sdk_readme_matrix.py + language: python + files: > + (?x) + ^java-sdk/capabilities\.yaml$| + ^java-sdk/gradle\.properties$| + ^java-sdk/README\.md$| + ^java-sdk/sdk/module\.md$| + ^scripts/ci/prek/lang_sdk_compat_matrix\.py$| + ^scripts/ci/prek/update_java_sdk_readme_matrix\.py$ + additional_dependencies: ['PyYAML>=6.0', 'rich>=13.6.0'] + pass_filenames: false + require_serial: true - id: check-go-version-in-sync name: Check Go toolchain version is consistent across build files entry: ./scripts/ci/prek/check_go_version_in_sync.py diff --git a/contributing-docs/30_new_language_sdk.rst b/contributing-docs/30_new_language_sdk.rst index 37b20b33dbd66..28928d0b4cca6 100644 --- a/contributing-docs/30_new_language_sdk.rst +++ b/contributing-docs/30_new_language_sdk.rst @@ -430,6 +430,11 @@ authored in the target language. An SDK declares each one independently. trailer (see `Native Executable Bundle Format`_), a JVM artifact embeds it in the jar, a Node bundle embeds it in the package. +``retry-policy`` (MAY) + The SDK lets task code inspect a failure and override whether the task retries or fails, + optionally with a custom retry delay. This is distinct from reporting ``up_for_retry``: + an SDK can support ordinary retries without exposing a task-facing retry-policy API. + ``task-state-store`` (MAY) The task can read and write the per-task state store. @@ -491,6 +496,30 @@ native Dags they are *not applicable* (``n/a``) rather than unsupported. The SDK exposes an object-storage API (an ``ObjectStoragePath`` equivalent) usable from native Dag code. +Compatibility matrix +~~~~~~~~~~~~~~~~~~~~ + +The dimensions above are prose; each SDK also declares them *machine-readably* in one +hand-authored ``/capabilities.yaml``, from which a prek hook generates its published tables: + +.. code-block:: text + + java-sdk/capabilities.yaml <- the only file you edit + | + | hook: update-java-sdk-readme-matrix + | + +--> java-sdk/README.md (contributor-facing) + +--> java-sdk/sdk/module.md (Dokka -> the published API reference) + +The hook rewrites its targets and exits non-zero when either was stale, so a drifted table fails +the build. Keep the manifest out of whatever the SDK publishes — it describes the SDK rather than +being part of it; for Java that means sitting above every subproject in ``settings.gradle.kts``. + +To add an SDK, register it in ``LANG_SDKS`` in ``scripts/ci/prek/lang_sdk_compat_matrix.py``, write +a ``capabilities.yaml`` in the same schema, and add the equivalent hook. Adding or renaming a +dimension means editing ``STATE_DIMENSIONS`` / ``CAPABILITY_DIMENSIONS`` there **and** the prose +above in the same PR — the renderer validates every manifest against that list. + Testing ------- diff --git a/java-sdk/README.md b/java-sdk/README.md index 64f53a5130c59..ef99848712760 100644 --- a/java-sdk/README.md +++ b/java-sdk/README.md @@ -580,6 +580,58 @@ Close the vote, **drop** the staging repository in Nexus, remove the `dist/dev` candidate, fix the issue, and cut the next RC (`...-rc2`). The released version stays the same (e.g. ``); only the RC counter in the tag increments. +## Compatibility matrix + +Which Airflow TaskInstance states and capabilities this SDK supports. This table is generated from +[`capabilities.yaml`](capabilities.yaml); the conformance dimensions are defined in the +[Language SDK conformance spec](https://github.com/apache/airflow/blob/main/contributing-docs/30_new_language_sdk.rst). +Do not edit the table by hand — edit `capabilities.yaml` and let the `update-java-sdk-readme-matrix` +prek hook regenerate it. + + + +*Min. Airflow version: 3.3 · supervisor schema: 2026-06-16* + +| Dimension | Tier | Supported | Since | Notes | +|---|---|---|---|---| +| **TaskInstance states** | | | | | +| state: `success` | MUST | ✓ | 3.3 | | +| state: `failed` | MUST | ✓ | 3.3 | | +| state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask | +| state: `skipped` | SHOULD | ✗ | – | runtime does not emit TaskState skipped yet | +| state: `deferred` | MAY | ✗ | – | runtime does not emit DeferTask yet | +| state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet | +| state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet | +| state: `removed` | MAY | ✓ | 3.3 | | +| **Runtime capabilities** | | | | | +| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.3 | @task.stub | +| capability: `task-logging` | MUST | ✓ | 3.3 | SLF4J + JPL bridged to the task log | +| capability: `xcom-read-write` | MUST | ✓ | 3.3 | | +| capability: `connection-read` | MUST | ✓ | 3.3 | | +| capability: `variable-read-write` | MUST | ✗ | – | getVariable only; no write over the comm socket yet | +| capability: `self-contained-bundle` | MUST | ✓ | 3.3 | Airflow metadata embedded in the jar artifact | +| capability: `retry-policy` | MAY | ✗ | – | no task-facing retry-policy API yet | +| capability: `task-state-store` | MAY | ✗ | – | no task-facing state-store API yet | +| capability: `asset-state-store` | MAY | ✗ | – | no task-facing state-store API yet | +| capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet | +| capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet | +| **Native-Dag authoring** | | | | | +| capability: `native-dag-authoring` | SHOULD | ✗ | – | native Dag authoring not implemented yet | +| capability: `task-args` | MUST † | n/a | – | | +| capability: `dag-params` | MUST † | n/a | – | | +| capability: `taskflow-dependencies` | MUST † | n/a | – | | +| capability: `branching` | SHOULD † | n/a | – | | +| capability: `dag-test` | SHOULD † | n/a | – | | +| capability: `task-group` | MAY † | n/a | – | | +| capability: `dynamic-task-mapping` | MAY † | n/a | – | | +| capability: `asset-inlets-outlets` | MAY † | n/a | – | | +| capability: `asset-scheduling` | MAY † | n/a | – | | +| capability: `object-store` | MAY † | n/a | – | | + +*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.* + + + ## Contributing The user implements a Java application containing task methods annotated (or diff --git a/java-sdk/capabilities.yaml b/java-sdk/capabilities.yaml new file mode 100644 index 0000000000000..a65a3f343c70f --- /dev/null +++ b/java-sdk/capabilities.yaml @@ -0,0 +1,116 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +--- +sdk: java + +min_airflow_version: "3.3" + +# Keep in sync with airflowSupervisorSchemaVersion in gradle.properties, which is what stamps the +# JAR manifest. The render hook fails if the two disagree. +supervisor_schema_version: "2026-06-16" + +# The runtime terminates a task with SucceedTask, RetryTask, or TaskState (failed/removed); it does +# not yet emit skipped, DeferTask, RescheduleTask, or AwaitInputTask. +states: + success: + supported: true + since: "3.3" + failed: + supported: true + since: "3.3" + up_for_retry: + supported: true + since: "3.3" + note: "RetryTask" + skipped: + supported: false + note: "runtime does not emit TaskState skipped yet" + deferred: + supported: false + note: "runtime does not emit DeferTask yet" + up_for_reschedule: + supported: false + note: "runtime does not emit RescheduleTask yet" + awaiting_input: + supported: false + note: "runtime does not emit AwaitInputTask yet" + removed: + supported: true + since: "3.3" + +# Runtime capabilities reflect the task-facing Client surface; native-Dag authoring is not +# implemented yet, so every native capability is unsupported. +capabilities: + mixed-lang-stub-target: + supported: true + since: "3.3" + note: "@task.stub" + task-logging: + supported: true + since: "3.3" + note: "SLF4J + JPL bridged to the task log" + xcom-read-write: + supported: true + since: "3.3" + connection-read: + supported: true + since: "3.3" + variable-read-write: + supported: false + note: "getVariable only; no write over the comm socket yet" + self-contained-bundle: + supported: true + since: "3.3" + note: "Airflow metadata embedded in the jar artifact" + retry-policy: + supported: false + note: "no task-facing retry-policy API yet" + task-state-store: + supported: false + note: "no task-facing state-store API yet" + asset-state-store: + supported: false + note: "no task-facing state-store API yet" + asset-event-emit: + supported: false + note: "runtime does not emit asset events yet" + asset-event-read: + supported: false + note: "no task-facing asset-event API yet" + native-dag-authoring: + supported: false + note: "native Dag authoring not implemented yet" + task-args: + supported: false + dag-params: + supported: false + taskflow-dependencies: + supported: false + branching: + supported: false + dag-test: + supported: false + task-group: + supported: false + dynamic-task-mapping: + supported: false + asset-inlets-outlets: + supported: false + asset-scheduling: + supported: false + object-store: + supported: false diff --git a/java-sdk/sdk/build.gradle.kts b/java-sdk/sdk/build.gradle.kts index e7d860569928b..060f37fc9e255 100644 --- a/java-sdk/sdk/build.gradle.kts +++ b/java-sdk/sdk/build.gradle.kts @@ -265,7 +265,15 @@ sourceSets { dokka { moduleVersion.set(project.version.toString()) + pluginsConfiguration.html { + // Widens the narrow compatibility-matrix columns; see the comments in the file. + customStyleSheets.from(layout.projectDirectory.file("dokka/matrix.css")) + } dokkaSourceSets.configureEach { + // Module-level documentation, including the generated Language SDK compatibility matrix. + // Dokka rejects the file unless "# Module sdk" is its very first line, so module.md carries + // the ASF license header just below the heading instead of above it. + includes.from("module.md") // Suppress everything in 'execution' since it's implementation detail. perPackageOption { matchingRegex = """org\.apache\.airflow\.sdk\.execution.*""" diff --git a/java-sdk/sdk/dokka/matrix.css b/java-sdk/sdk/dokka/matrix.css new file mode 100644 index 0000000000000..6396b7ff028a2 --- /dev/null +++ b/java-sdk/sdk/dokka/matrix.css @@ -0,0 +1,31 @@ +/*! + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +/* Tier ("SHOULD †"), Supported ("n/a") and Since ("3.3") in the compatibility matrix. */ +.table--container th:nth-child(2), +.table--container th:nth-child(3), +.table--container td:nth-child(2), +.table--container td:nth-child(3) { + min-width: 9ch; +} + +.table--container th:nth-child(4), +.table--container td:nth-child(4) { + min-width: 6ch; +} diff --git a/java-sdk/sdk/module.md b/java-sdk/sdk/module.md new file mode 100644 index 0000000000000..b1189de7cbe70 --- /dev/null +++ b/java-sdk/sdk/module.md @@ -0,0 +1,72 @@ +# Module sdk + + + +The Apache Airflow Java SDK — author and run Airflow task implementations in JVM languages. + +## Language SDK compatibility matrix + +Which Airflow TaskInstance states and capabilities the Java SDK currently supports. The normative +meaning of each dimension is defined in the +[Language SDK conformance specification](https://github.com/apache/airflow/blob/main/contributing-docs/30_new_language_sdk.rst). + + + +*Min. Airflow version: 3.3 · supervisor schema: 2026-06-16* + +| Dimension | Tier | Supported | Since | Notes | +|---|---|---|---|---| +| **TaskInstance states** | | | | | +| state: `success` | MUST | ✓ | 3.3 | | +| state: `failed` | MUST | ✓ | 3.3 | | +| state: `up_for_retry` | MUST | ✓ | 3.3 | RetryTask | +| state: `skipped` | SHOULD | ✗ | – | runtime does not emit TaskState skipped yet | +| state: `deferred` | MAY | ✗ | – | runtime does not emit DeferTask yet | +| state: `up_for_reschedule` | MAY | ✗ | – | runtime does not emit RescheduleTask yet | +| state: `awaiting_input` | MAY | ✗ | – | runtime does not emit AwaitInputTask yet | +| state: `removed` | MAY | ✓ | 3.3 | | +| **Runtime capabilities** | | | | | +| capability: `mixed-lang-stub-target` | MUST | ✓ | 3.3 | @task.stub | +| capability: `task-logging` | MUST | ✓ | 3.3 | SLF4J + JPL bridged to the task log | +| capability: `xcom-read-write` | MUST | ✓ | 3.3 | | +| capability: `connection-read` | MUST | ✓ | 3.3 | | +| capability: `variable-read-write` | MUST | ✗ | – | getVariable only; no write over the comm socket yet | +| capability: `self-contained-bundle` | MUST | ✓ | 3.3 | Airflow metadata embedded in the jar artifact | +| capability: `retry-policy` | MAY | ✗ | – | no task-facing retry-policy API yet | +| capability: `task-state-store` | MAY | ✗ | – | no task-facing state-store API yet | +| capability: `asset-state-store` | MAY | ✗ | – | no task-facing state-store API yet | +| capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet | +| capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet | +| **Native-Dag authoring** | | | | | +| capability: `native-dag-authoring` | SHOULD | ✗ | – | native Dag authoring not implemented yet | +| capability: `task-args` | MUST † | n/a | – | | +| capability: `dag-params` | MUST † | n/a | – | | +| capability: `taskflow-dependencies` | MUST † | n/a | – | | +| capability: `branching` | SHOULD † | n/a | – | | +| capability: `dag-test` | SHOULD † | n/a | – | | +| capability: `task-group` | MAY † | n/a | – | | +| capability: `dynamic-task-mapping` | MAY † | n/a | – | | +| capability: `asset-inlets-outlets` | MAY † | n/a | – | | +| capability: `asset-scheduling` | MAY † | n/a | – | | +| capability: `object-store` | MAY † | n/a | – | | + +*Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.* + + diff --git a/scripts/ci/prek/lang_sdk_compat_matrix.py b/scripts/ci/prek/lang_sdk_compat_matrix.py new file mode 100644 index 0000000000000..7de1b3a30a9a8 --- /dev/null +++ b/scripts/ci/prek/lang_sdk_compat_matrix.py @@ -0,0 +1,305 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Shared helpers for the Language SDK compatibility matrix. + +Every Language SDK (Go, Java, TypeScript) declares what it supports in a hand-authored +``/capabilities.yaml`` at the root of its own tree. This module owns the *schema* of that +file, the *registry* of SDKs, and :func:`render_markdown_table`, which renders the per-SDK Markdown +table that each SDK's own prek hook embeds in its docs. + +Because the manifest is hand-authored, :func:`validate_capabilities` is the only thing standing +between a typo and a wrong published table, so it rejects unknown keys as well as missing ones. + +The normative meaning of each dimension lives in ``contributing-docs/30_new_language_sdk.rst`` +(the "Conformance" section). Keep the dimensions below in sync with that document. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import NamedTuple, TypedDict + +import yaml +from common_prek_utils import AIRFLOW_ROOT_PATH + +# Markers wrapping the generated tables. insert_documentation() keeps these lines and rewrites +# everything between them, so the same constants are reused by every SDK's README hook. +README_MATRIX_HEADER = "" +README_MATRIX_FOOTER = "" + +SUPPORTED_MARK = "✓" +UNSUPPORTED_MARK = "✗" +NA_MARK = "n/a" # a gated native-Dag capability while native-dag-authoring is unsupported +NO_VERSION_MARK = "–" # "Since" placeholder for a dimension that is not supported + +# The umbrella capability that gates the conditional native-Dag capabilities: when an SDK does +# not support it, every gated native-Dag capability is "not applicable" rather than unsupported. +NATIVE_DAG_GATE = "native-dag-authoring" + +# TaskInstance states a subprocess can emit, in display order, with their conformance tier. +# Scheduler-owned states (queued, scheduled, running, restarting, upstream_failed) are never +# emitted by an SDK runtime and are deliberately excluded. +STATE_DIMENSIONS: list[tuple[str, str]] = [ + ("success", "MUST"), + ("failed", "MUST"), + ("up_for_retry", "MUST"), + ("skipped", "SHOULD"), + ("deferred", "MAY"), + ("up_for_reschedule", "MAY"), + ("awaiting_input", "MAY"), + ("removed", "MAY"), +] + + +class Capability(NamedTuple): + name: str + tier: str + group: str # "runtime" or "native" + gated: bool # renders n/a (not ✗) when NATIVE_DAG_GATE is unsupported + + +# Capability flags, in display order. Runtime capabilities describe what a task body can do while +# it runs (in either a mixed-lang or native Dag); native-Dag capabilities describe authoring a +# whole Dag in the target language and are gated by NATIVE_DAG_GATE (except the gate itself). +CAPABILITY_DIMENSIONS: list[Capability] = [ + Capability("mixed-lang-stub-target", "MUST", "runtime", False), + Capability("task-logging", "MUST", "runtime", False), + Capability("xcom-read-write", "MUST", "runtime", False), + Capability("connection-read", "MUST", "runtime", False), + Capability("variable-read-write", "MUST", "runtime", False), + Capability("self-contained-bundle", "MUST", "runtime", False), + Capability("retry-policy", "MAY", "runtime", False), + Capability("task-state-store", "MAY", "runtime", False), + Capability("asset-state-store", "MAY", "runtime", False), + Capability("asset-event-emit", "MAY", "runtime", False), + Capability("asset-event-read", "MAY", "runtime", False), + Capability(NATIVE_DAG_GATE, "SHOULD", "native", False), + Capability("task-args", "MUST", "native", True), + Capability("dag-params", "MUST", "native", True), + Capability("taskflow-dependencies", "MUST", "native", True), + Capability("branching", "SHOULD", "native", True), + Capability("dag-test", "SHOULD", "native", True), + Capability("task-group", "MAY", "native", True), + Capability("dynamic-task-mapping", "MAY", "native", True), + Capability("asset-inlets-outlets", "MAY", "native", True), + Capability("asset-scheduling", "MAY", "native", True), + Capability("object-store", "MAY", "native", True), +] + +CAPABILITY_NAMES = {cap.name for cap in CAPABILITY_DIMENSIONS} + +GROUP_LABELS = {"runtime": "Runtime capabilities", "native": "Native-Dag authoring"} +STATES_GROUP_LABEL = "TaskInstance states" + +LEGEND = ( + f"Marks: {SUPPORTED_MARK} supported · {UNSUPPORTED_MARK} not supported · " + f"{NA_MARK} not applicable. A tier marked † applies only when `{NATIVE_DAG_GATE}` is supported." +) + + +class LangSdk(TypedDict): + id: str + capabilities_yaml: Path + readme: Path + + +# The registry of Language SDKs and where each one's manifest and README live. Only the Java SDK +# declares one so far; the Go and TypeScript entries record where theirs go when those runtimes +# declare their capabilities. Because of that, `capabilities_yaml` is a declared location and not a +# promise the file exists — a consumer walking the whole registry must check `.exists()` before +# calling load_capabilities(). +LANG_SDKS: list[LangSdk] = [ + { + "id": "go", + "capabilities_yaml": AIRFLOW_ROOT_PATH / "go-sdk" / "capabilities.yaml", + "readme": AIRFLOW_ROOT_PATH / "go-sdk" / "README.md", + }, + { + "id": "java", + "capabilities_yaml": AIRFLOW_ROOT_PATH / "java-sdk" / "capabilities.yaml", + "readme": AIRFLOW_ROOT_PATH / "java-sdk" / "README.md", + }, + { + "id": "ts", + "capabilities_yaml": AIRFLOW_ROOT_PATH / "ts-sdk" / "capabilities.yaml", + "readme": AIRFLOW_ROOT_PATH / "ts-sdk" / "README.md", + }, +] + +VALID_SDK_IDS = {sdk["id"] for sdk in LANG_SDKS} + + +class DimensionEntry(TypedDict, total=False): + supported: bool + since: str | None + note: str + + +class CapabilitiesDoc(TypedDict): + sdk: str + supervisor_schema_version: str + min_airflow_version: str + states: dict[str, DimensionEntry] + capabilities: dict[str, DimensionEntry] + + +class CapabilitiesError(ValueError): + """Raised when a capabilities.yaml file does not match the expected schema.""" + + +def load_capabilities(path: Path, *, expected_sdk: str | None = None) -> CapabilitiesDoc: + """Load and validate a ``capabilities.yaml`` file. + + ``expected_sdk`` binds the file to the SDK it belongs to: passing it makes a manifest whose + ``sdk`` field disagrees with the file's own SDK (e.g. ``go-sdk/capabilities.yaml`` declaring + ``sdk: java``) a validation error instead of silently rendering in the wrong column. + """ + doc = yaml.safe_load(path.read_text()) + validate_capabilities(doc, source=str(path), expected_sdk=expected_sdk) + return doc + + +def validate_capabilities(doc: object, *, source: str, expected_sdk: str | None = None) -> None: + """Validate a decoded capabilities document, raising :class:`CapabilitiesError` on any issue. + + When ``expected_sdk`` is given, the document's ``sdk`` field must equal it. + """ + if not isinstance(doc, dict): + raise CapabilitiesError(f"{source}: top-level value must be a mapping") + required = {"sdk", "supervisor_schema_version", "min_airflow_version", "states", "capabilities"} + missing = required - doc.keys() + if missing: + raise CapabilitiesError(f"{source}: missing required keys: {', '.join(sorted(missing))}") + unknown = doc.keys() - required + if unknown: + raise CapabilitiesError(f"{source}: unknown top-level keys: {', '.join(sorted(unknown))}") + if doc["sdk"] not in VALID_SDK_IDS: + raise CapabilitiesError( + f"{source}: unknown sdk {doc['sdk']!r}; expected one of {sorted(VALID_SDK_IDS)}" + ) + if expected_sdk is not None and doc["sdk"] != expected_sdk: + raise CapabilitiesError(f"{source}: sdk is {doc['sdk']!r} but this file belongs to {expected_sdk!r}") + for field in ("supervisor_schema_version", "min_airflow_version"): + if not isinstance(doc[field], str): + raise CapabilitiesError(f"{source}: {field} must be a string") + _validate_entries( + doc["states"], expected={state for state, _ in STATE_DIMENSIONS}, kind="states", source=source + ) + _validate_entries(doc["capabilities"], expected=CAPABILITY_NAMES, kind="capabilities", source=source) + if not doc["capabilities"][NATIVE_DAG_GATE]["supported"]: + supported_gated = sorted( + cap.name + for cap in CAPABILITY_DIMENSIONS + if cap.gated and doc["capabilities"][cap.name]["supported"] + ) + if supported_gated: + raise CapabilitiesError( + f"{source}: gated capabilities cannot be supported while {NATIVE_DAG_GATE!r} " + f"is not supported: {', '.join(supported_gated)}" + ) + + +def _validate_entries(entries: object, *, expected: set[str], kind: str, source: str) -> None: + if not isinstance(entries, dict): + raise CapabilitiesError(f"{source}: {kind!r} must be a mapping") + actual = set(entries.keys()) + if actual != expected: + missing = expected - actual + unknown = actual - expected + problems = [] + if missing: + problems.append(f"missing {sorted(missing)}") + if unknown: + problems.append(f"unknown {sorted(unknown)}") + raise CapabilitiesError(f"{source}: {kind} keys mismatch: {'; '.join(problems)}") + for name, entry in entries.items(): + if not isinstance(entry, dict) or not isinstance(entry.get("supported"), bool): + raise CapabilitiesError(f"{source}: {kind}.{name} must be a mapping with a boolean 'supported'") + # A misspelled optional key would otherwise be dropped silently and render as a blank cell. + unknown_fields = entry.keys() - {"supported", "since", "note"} + if unknown_fields: + raise CapabilitiesError( + f"{source}: {kind}.{name} has unknown keys: {', '.join(sorted(unknown_fields))}" + ) + if not isinstance(entry.get("since", None), (str, type(None))): + raise CapabilitiesError(f"{source}: {kind}.{name}.since must be a string or null") + if not entry["supported"] and entry.get("since") is not None: + # "Since" means "supported since"; carrying one while unsupported is contradictory and + # would silently render as the not-supported placeholder. Supported *without* a version + # stays legal — an SDK may not know which release first shipped a dimension. + raise CapabilitiesError( + f"{source}: {kind}.{name} is not supported but carries since=" + f"{entry['since']!r}; drop the version or mark it supported" + ) + if not isinstance(entry.get("note", ""), str): + raise CapabilitiesError(f"{source}: {kind}.{name}.note must be a string") + + +def _state_mark(entry: DimensionEntry) -> str: + return SUPPORTED_MARK if entry.get("supported") else UNSUPPORTED_MARK + + +def _capability_mark(doc: CapabilitiesDoc, cap: Capability) -> str: + if cap.gated and not doc["capabilities"][NATIVE_DAG_GATE].get("supported"): + return NA_MARK + return SUPPORTED_MARK if doc["capabilities"][cap.name].get("supported") else UNSUPPORTED_MARK + + +def _since(entry: DimensionEntry) -> str: + if not entry.get("supported"): + return NO_VERSION_MARK + return entry.get("since") or NO_VERSION_MARK + + +def _note(entry: DimensionEntry) -> str: + return (entry.get("note") or "").replace("|", "\\|") + + +def _tier_label(cap: Capability) -> str: + return f"{cap.tier} †" if cap.gated else cap.tier + + +def render_markdown_table(doc: CapabilitiesDoc) -> list[str]: + """Render the per-SDK Markdown compatibility table as a list of lines (trailing newlines).""" + lines = [ + "\n", + f"*Min. Airflow version: {doc['min_airflow_version']} · " + f"supervisor schema: {doc['supervisor_schema_version']}*\n", + "\n", + "| Dimension | Tier | Supported | Since | Notes |\n", + "|---|---|---|---|---|\n", + f"| **{STATES_GROUP_LABEL}** | | | | |\n", + ] + for state, tier in STATE_DIMENSIONS: + entry = doc["states"][state] + lines.append( + f"| state: `{state}` | {tier} | {_state_mark(entry)} | {_since(entry)} | {_note(entry)} |\n" + ) + current_group = "" + for cap in CAPABILITY_DIMENSIONS: + if cap.group != current_group: + current_group = cap.group + lines.append(f"| **{GROUP_LABELS[cap.group]}** | | | | |\n") + entry = doc["capabilities"][cap.name] + lines.append( + f"| capability: `{cap.name}` | {_tier_label(cap)} | {_capability_mark(doc, cap)} | " + f"{_since(entry)} | {_note(entry)} |\n" + ) + lines.append("\n") + lines.append(f"*{LEGEND}*\n") + lines.append("\n") + return lines diff --git a/scripts/ci/prek/update_java_sdk_readme_matrix.py b/scripts/ci/prek/update_java_sdk_readme_matrix.py new file mode 100755 index 0000000000000..2284d0dd45b03 --- /dev/null +++ b/scripts/ci/prek/update_java_sdk_readme_matrix.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# /// script +# requires-python = ">=3.10,<3.11" +# dependencies = ["PyYAML>=6.0", "rich>=13.6.0"] +# /// +"""Regenerate the Java SDK compatibility table in ``java-sdk/README.md`` and the Dokka module doc. + +Renders the Markdown matrix from ``java-sdk/capabilities.yaml`` between the AUTO-GENERATED markers +in both ``java-sdk/README.md`` and ``java-sdk/sdk/module.md`` (the latter is included in the Dokka +API reference via ``includes.from("module.md")``). Exits non-zero when either file changed so the +contributor re-stages it. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from common_prek_utils import console, insert_documentation +from lang_sdk_compat_matrix import ( + AIRFLOW_ROOT_PATH, + LANG_SDKS, + README_MATRIX_FOOTER, + README_MATRIX_HEADER, + CapabilitiesDoc, + load_capabilities, + render_markdown_table, +) + +SDK_ID = "java" + +# The Dokka module documentation that surfaces the matrix in the Java API reference. +DOKKA_MODULE_DOC = AIRFLOW_ROOT_PATH / "java-sdk" / "sdk" / "module.md" + +GRADLE_PROPERTIES = AIRFLOW_ROOT_PATH / "java-sdk" / "gradle.properties" +SCHEMA_VERSION_PROPERTY = "airflowSupervisorSchemaVersion" + + +def read_gradle_schema_version() -> str | None: + """The ``airflowSupervisorSchemaVersion`` from gradle.properties (source of truth for the JAR).""" + for line in GRADLE_PROPERTIES.read_text().splitlines(): + key, sep, value = line.partition("=") + if sep and key.strip() == SCHEMA_VERSION_PROPERTY: + return value.strip() + return None + + +def check_schema_version(doc: CapabilitiesDoc) -> bool: + """Whether the manifest agrees with the schema version the JAR manifest is stamped with.""" + gradle_version = read_gradle_schema_version() + declared = doc["supervisor_schema_version"] + if gradle_version is None or gradle_version == declared: + return True + console.print( + f"[red]java-sdk/capabilities.yaml declares supervisor_schema_version {declared!r} but " + f"gradle.properties {SCHEMA_VERSION_PROPERTY} is {gradle_version!r} (the JAR manifest uses " + f"the latter). Update capabilities.yaml to match.[/]" + ) + return False + + +def main() -> int: + sdk = next(entry for entry in LANG_SDKS if entry["id"] == SDK_ID) + doc = load_capabilities(sdk["capabilities_yaml"], expected_sdk=SDK_ID) + if not check_schema_version(doc): + return 1 + table = render_markdown_table(doc) + changed = False + for target, label in ( + (sdk["readme"], "java-sdk/README.md"), + (DOKKA_MODULE_DOC, "java-sdk/sdk/module.md"), + ): + if insert_documentation( + target, + table, + README_MATRIX_HEADER, + README_MATRIX_FOOTER, + extra_information="the Java SDK compatibility matrix", + ): + console.print( + f"[yellow]Regenerated the Java SDK compatibility matrix in {label}; re-stage it.[/]" + ) + changed = True + return 1 if changed else 0 + + +if __name__ in ("__main__", "__mp_main__"): + raise SystemExit(main()) diff --git a/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py b/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py new file mode 100644 index 0000000000000..38a0a29f8f235 --- /dev/null +++ b/scripts/tests/ci/prek/test_lang_sdk_compat_matrix.py @@ -0,0 +1,203 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import pytest +import yaml +from ci.prek import lang_sdk_compat_matrix as matrix + + +def _entry(supported: bool, since: str | None = None, note: str = "") -> dict: + return {"supported": supported, "since": since, "note": note} + + +def _doc(sdk_id: str = "go", **overrides) -> dict: + """A fully-populated, valid capabilities document for ``sdk_id``.""" + doc = { + "sdk": sdk_id, + "supervisor_schema_version": "2026-06-16", + "min_airflow_version": "3.3", + "states": {state: _entry(True, since="3.3") for state, _ in matrix.STATE_DIMENSIONS}, + "capabilities": {cap.name: _entry(True, since="3.3") for cap in matrix.CAPABILITY_DIMENSIONS}, + } + doc.update(overrides) + return doc + + +class TestValidateCapabilities: + def test_valid_doc_passes(self): + matrix.validate_capabilities(_doc(), source="test") + + def test_non_dict_raises(self): + with pytest.raises(matrix.CapabilitiesError, match="must be a mapping"): + matrix.validate_capabilities([], source="test") + + @pytest.mark.parametrize( + "key", ["sdk", "supervisor_schema_version", "min_airflow_version", "states", "capabilities"] + ) + def test_missing_required_key_raises(self, key): + doc = _doc() + del doc[key] + with pytest.raises(matrix.CapabilitiesError, match="missing required keys"): + matrix.validate_capabilities(doc, source="test") + + def test_unknown_top_level_key_raises(self): + with pytest.raises(matrix.CapabilitiesError, match="unknown top-level keys: capability"): + matrix.validate_capabilities(_doc(capability={}), source="test") + + def test_unknown_entry_key_raises(self): + doc = _doc() + doc["capabilities"]["branching"] = {"supported": True, "supported_since": "3.3"} + with pytest.raises(matrix.CapabilitiesError, match="branching has unknown keys: supported_since"): + matrix.validate_capabilities(doc, source="test") + + def test_unknown_sdk_raises(self): + with pytest.raises(matrix.CapabilitiesError, match="unknown sdk"): + matrix.validate_capabilities(_doc(sdk_id="rust"), source="test") + + def test_missing_state_raises(self): + doc = _doc() + del doc["states"]["deferred"] + with pytest.raises(matrix.CapabilitiesError, match="states keys mismatch"): + matrix.validate_capabilities(doc, source="test") + + def test_unknown_capability_raises(self): + doc = _doc() + doc["capabilities"]["telepathy"] = _entry(True) + with pytest.raises(matrix.CapabilitiesError, match="capabilities keys mismatch"): + matrix.validate_capabilities(doc, source="test") + + def test_non_boolean_supported_raises(self): + doc = _doc() + doc["states"]["success"] = {"supported": "true"} + with pytest.raises(matrix.CapabilitiesError, match="boolean 'supported'"): + matrix.validate_capabilities(doc, source="test") + + def test_sdk_mismatch_against_expected_raises(self): + with pytest.raises(matrix.CapabilitiesError, match="belongs to 'go'"): + matrix.validate_capabilities(_doc(sdk_id="java"), source="test", expected_sdk="go") + + def test_expected_sdk_match_passes(self): + matrix.validate_capabilities(_doc(sdk_id="go"), source="test", expected_sdk="go") + + def test_non_string_note_raises(self): + doc = _doc() + doc["states"]["success"] = {"supported": True, "since": "3.3", "note": 123} + with pytest.raises(matrix.CapabilitiesError, match="note must be a string"): + matrix.validate_capabilities(doc, source="test") + + def test_non_string_since_raises(self): + doc = _doc() + doc["capabilities"]["xcom-read-write"] = {"supported": True, "since": 3, "note": ""} + with pytest.raises(matrix.CapabilitiesError, match="since must be a string or null"): + matrix.validate_capabilities(doc, source="test") + + def test_non_string_schema_version_raises(self): + with pytest.raises(matrix.CapabilitiesError, match="supervisor_schema_version must be a string"): + matrix.validate_capabilities(_doc(supervisor_schema_version=123), source="test") + + def test_unsupported_entry_carrying_since_raises(self): + doc = _doc() + doc["capabilities"]["branching"] = _entry(False, since="3.3") + with pytest.raises(matrix.CapabilitiesError, match="not supported but carries since"): + matrix.validate_capabilities(doc, source="test") + + def test_supported_gated_capability_without_native_dag_authoring_raises(self): + doc = _doc() + doc["capabilities"][matrix.NATIVE_DAG_GATE] = _entry(False) + with pytest.raises( + matrix.CapabilitiesError, + match="gated capabilities cannot be supported.*branching", + ): + matrix.validate_capabilities(doc, source="test") + + def test_supported_entry_without_since_passes(self): + doc = _doc() + doc["states"]["success"] = _entry(True, since=None) + matrix.validate_capabilities(doc, source="test") + + def test_entry_omitting_optional_keys_passes(self): + doc = _doc() + doc["states"]["success"] = {"supported": True} + matrix.validate_capabilities(doc, source="test") + + +class TestLoadCapabilities: + def test_reads_yaml(self, tmp_path): + path = tmp_path / "capabilities.yaml" + path.write_text(yaml.safe_dump(_doc())) + assert matrix.load_capabilities(path, expected_sdk="go") == _doc() + + def test_committed_manifests_are_valid(self): + """Every manifest an SDK has actually committed passes validation.""" + declared = [sdk for sdk in matrix.LANG_SDKS if sdk["capabilities_yaml"].exists()] + assert declared, "expected at least one Language SDK to declare its capabilities" + for sdk in declared: + matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=sdk["id"]) + + +class TestRenderMarkdownTable: + def test_retry_policy_is_an_optional_runtime_capability(self): + retry_policy = next(cap for cap in matrix.CAPABILITY_DIMENSIONS if cap.name == "retry-policy") + assert retry_policy == matrix.Capability("retry-policy", "MAY", "runtime", False) + + def test_rows_cover_every_dimension(self): + rendered = "".join(matrix.render_markdown_table(_doc())) + for state, tier in matrix.STATE_DIMENSIONS: + assert f"| state: `{state}` | {tier} |" in rendered + for cap in matrix.CAPABILITY_DIMENSIONS: + assert f"| capability: `{cap.name}` | {matrix._tier_label(cap)} |" in rendered + assert "supervisor schema: 2026-06-16" in rendered + + def test_supported_and_unsupported_marks(self): + doc = _doc() + doc["states"]["deferred"] = _entry(False, note="no triggerer bridge") + rendered = "".join(matrix.render_markdown_table(doc)) + assert f"| state: `success` | MUST | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered + assert ( + f"| state: `deferred` | MAY | {matrix.UNSUPPORTED_MARK} | {matrix.NO_VERSION_MARK} |" in rendered + ) + + def test_gated_capabilities_are_na_without_the_native_dag_gate(self): + doc = _doc() + doc["capabilities"][matrix.NATIVE_DAG_GATE] = _entry(False) + rendered = "".join(matrix.render_markdown_table(doc)) + assert f"| capability: `{matrix.NATIVE_DAG_GATE}` | SHOULD | {matrix.UNSUPPORTED_MARK} |" in rendered + for cap in matrix.CAPABILITY_DIMENSIONS: + if cap.gated: + assert f"| capability: `{cap.name}` | {cap.tier} † | {matrix.NA_MARK} |" in rendered + + def test_gated_capabilities_resolve_to_their_own_mark_once_the_gate_is_supported(self): + doc = _doc() + doc["capabilities"]["branching"] = _entry(False, note="no branch construct") + rendered = "".join(matrix.render_markdown_table(doc)) + gated = [cap for cap in matrix.CAPABILITY_DIMENSIONS if cap.gated] + assert gated, "expected at least one gated capability" + # With the gate supported, a gated capability reports its real support, never n/a. + for cap in gated: + if cap.name != "branching": + assert ( + f"| capability: `{cap.name}` | {cap.tier} † | {matrix.SUPPORTED_MARK} | 3.3 |" in rendered + ) + assert f"| capability: `branching` | SHOULD † | {matrix.UNSUPPORTED_MARK} |" in rendered + assert matrix.NA_MARK not in rendered.replace(matrix.LEGEND, "") + + def test_pipe_in_note_is_escaped(self): + doc = _doc() + doc["capabilities"]["xcom-read-write"] = _entry(True, since="3.3", note="read | write") + rendered = "".join(matrix.render_markdown_table(doc)) + assert "read \\| write" in rendered diff --git a/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py b/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py new file mode 100644 index 0000000000000..730f030c64a40 --- /dev/null +++ b/scripts/tests/ci/prek/test_update_java_sdk_readme_matrix.py @@ -0,0 +1,86 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import pytest +import yaml +from ci.prek import lang_sdk_compat_matrix as matrix, update_java_sdk_readme_matrix as hook + +SCHEMA_VERSION = "2026-06-16" + + +def _doc() -> dict: + def entry(supported: bool) -> dict: + return {"supported": supported, "since": "3.3" if supported else None, "note": ""} + + return { + "sdk": "java", + "supervisor_schema_version": SCHEMA_VERSION, + "min_airflow_version": "3.3", + "states": {state: entry(True) for state, _ in matrix.STATE_DIMENSIONS}, + "capabilities": {cap.name: entry(True) for cap in matrix.CAPABILITY_DIMENSIONS}, + } + + +class TestMain: + @pytest.fixture + def wired(self, tmp_path, monkeypatch): + """Point the hook at a temp capabilities.yaml plus temp README and Dokka module doc.""" + capabilities_yaml = tmp_path / "capabilities.yaml" + capabilities_yaml.write_text(yaml.safe_dump(_doc())) + gradle_properties = tmp_path / "gradle.properties" + gradle_properties.write_text( + f"projectVersion=1.0.0-SNAPSHOT\n{hook.SCHEMA_VERSION_PROPERTY}={SCHEMA_VERSION}\n" + ) + targets = [] + for name in ("README.md", "module.md"): + target = tmp_path / name + target.write_text( + f"intro\n\n{matrix.README_MATRIX_HEADER}\n{matrix.README_MATRIX_FOOTER}\n\noutro\n" + ) + targets.append(target) + readme, module_doc = targets + monkeypatch.setattr( + hook, + "LANG_SDKS", + [{"id": "java", "capabilities_yaml": capabilities_yaml, "readme": readme}], + ) + monkeypatch.setattr(hook, "DOKKA_MODULE_DOC", module_doc) + monkeypatch.setattr(hook, "GRADLE_PROPERTIES", gradle_properties) + return readme, module_doc + + def test_generates_both_targets_then_is_idempotent(self, wired): + assert hook.main() == 1 + for target in wired: + content = target.read_text() + assert "| Dimension | Tier | Supported | Since | Notes |" in content + assert matrix.SUPPORTED_MARK in content + assert content.startswith("intro\n") and content.endswith("outro\n") + + assert hook.main() == 0 + + def test_schema_version_disagreeing_with_gradle_fails_without_writing(self, wired): + hook.GRADLE_PROPERTIES.write_text(f"{hook.SCHEMA_VERSION_PROPERTY}=2020-01-01\n") + assert hook.main() == 1 + for target in wired: + assert matrix.README_MATRIX_HEADER + "\n" + matrix.README_MATRIX_FOOTER in target.read_text() + + def test_schema_version_matches_gradle_properties(self): + """The committed manifest agrees with the version that stamps the JAR.""" + sdk = next(entry for entry in matrix.LANG_SDKS if entry["id"] == hook.SDK_ID) + doc = matrix.load_capabilities(sdk["capabilities_yaml"], expected_sdk=hook.SDK_ID) + assert doc["supervisor_schema_version"] == hook.read_gradle_schema_version()