Java SDK: Serialize native Dags to DagSerialization v3

.pre-commit-config.yaml

@@ -430,6 +430,25 @@ repos:
           ^go-sdk/schema/dag-schema\.json$|
           ^scripts/ci/lang_sdk_serialization/.*$|
           ^scripts/ci/prek/check_go_sdk_serialization_conformance\.py$
+      - id: check-java-sdk-serialization-conformance
+        name: Check the Java SDK serializes Dags the way Airflow does
+        description: "Serialize the shared test Dags with the Java SDK and with Airflow, and compare the two"
+        entry: ./java-sdk/scripts/ci/prek/check_serialization_conformance.py
+        language: system
+        pass_filenames: false
+        require_serial: true
+        files: >
+          (?x)
+          ^airflow-core/src/airflow/serialization/serialized_objects\.py$|
+          ^airflow-core/src/airflow/serialization/schema\.json$|
+          ^java-sdk/scripts/ci/prek/check_serialization_conformance\.py$|
+          ^java-sdk/sdk/schema/dag-schema\.json$|
+          ^java-sdk/sdk/build\.gradle\.kts$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/(Bundle|DagDef|Deps|Arg|TaskGroupRef)\.kt$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde\.kt$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/(Refs|Fields)\.kt$|
+          ^java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/conformance/.*$|
+          ^scripts/ci/lang_sdk_serialization/.*$
       - 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

airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md

@@ -256,6 +256,11 @@ modes. It is also why the spec is deliberately not `@task.stub`-shaped — no fi
 natively authored call. (Its docstrings still say "stub", reflecting the only producer that
 exists today; that is wording to revisit, not a constraint in the format.)
 
+A runtime that holds its own Dag may resolve call arguments from that Dag rather than from the
+delivered spec. The spec stays the recorded and displayed form, which keeps one wire format across
+both authoring modes. The Go SDK reads the delivered bindings and the Java SDK resolves locally;
+both emit the same spec.
+
 ### H. Scope: what does not cross the boundary
 
 The following raise when the Dag is serialized, rather than being silently dropped or deferred to

airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst

@@ -675,7 +675,9 @@ generated view, with a no-argument ``depends()`` method:
 
 Every ``@Builder.Task`` method must be called in the wiring class; a task the wiring missed fails at
 Dag-parse time.  ``lit(...)`` wires an inline constant where no upstream feeds a parameter.  A bare
-``double`` cannot be an ``Arg``, so a constant is wrapped.  A view method that takes no arguments
+``double`` cannot be an ``Arg``, so a constant is wrapped.  Airflow records what each task is called
+with, and that record travels as JSON, so a constant has to be a string, number, boolean, list or
+map.  A view method that takes no arguments
 returns the same handle every time, so it names one node wherever it appears; one that takes
 arguments is called once, and the wiring fails if it is called again with arguments, so hold its
 handle in a local and reuse that.
@@ -690,9 +692,9 @@ class that supplies only task bodies, for a Dag a Python file declares, carries
 
 .. note::
 
-   Runtime argument bindings win over Java-declared wiring.  When the supervisor delivers bindings
-   for a run (see :ref:`java-sdk/arg-binding`), the binding at a parameter's position is what the
-   task receives.  Wired inputs are the fallback, which is what a native Java Dag always uses.
+   A native Java Dag binds its task arguments from its own wiring, and the ``_arg_bindings`` it
+   serializes are what Airflow records and shows.  Runtime bindings (see :ref:`java-sdk/arg-binding`)
+   are what a ``@Builder.TaskHandler`` class reads, for a task whose Dag a Python file declares.
 
 Task groups
 ~~~~~~~~~~~
@@ -772,6 +774,9 @@ Durations and date-times are ISO-8601 strings in annotations (``retryDelay = "PT
 ``java.time.OffsetDateTime`` values in ``config`` calls.  An unknown key or a mismatched value type
 fails the build for an annotation, and the ``config`` call itself for an object.
 
+A Dag with a cron ``schedule`` runs in the time zone of its ``startDate``.  With no ``startDate`` it
+is scheduled in UTC, so set ``startDate`` to pin the zone.
+
 .. _java-sdk/task-state-store:
 
 Task state store

dev/breeze/doc/ci/04_selective_checks.md

@@ -548,6 +548,12 @@ when some files are not changed. Those are the rules implemented:
     Gradle wrapper, which downloads the Gradle distribution, and the latter additionally
     resolves the whole Java SDK dependency graph from Maven Central, so we avoid those
     downloads on PRs that do not touch `java-sdk/`)
+  * if no `Java SDK conformance files` changed - `check-java-sdk-serialization-conformance`
+    is skipped (it compiles the Java SDK with Gradle and resolves its dependencies from Maven
+    Central). The group is wider than `Java SDK files`: it also covers Airflow's serializer
+    (`serialized_objects.py`), `schema.json` and the shared harness under
+    `scripts/ci/lang_sdk_serialization/`, because the check compares the SDK against those and
+    none of them forces `full_tests_needed`
   * if no `TS SDK files` (`ts-sdk/`) changed - `check-ts-sdk-supervisor-schema` check is
     skipped (it regenerates and diffs the generated ts-sdk file; a change to the supervisor
     wire schema alone deliberately does not trigger it - regenerating the ts-sdk types is

dev/breeze/src/airflow_breeze/utils/selective_checks.py

@@ -140,6 +140,7 @@ class FileGroupForCi(Enum):
     AGENT_FRAMEWORK_FILES = auto()
     GO_SDK_FILES = auto()
     JAVA_SDK_FILES = auto()
+    JAVA_SDK_CONFORMANCE_FILES = auto()
     TS_SDK_FILES = auto()
     TS_SDK_DOCS_FILES = auto()
     AIRFLOW_CTL_FILES = auto()
@@ -517,6 +518,14 @@ def __hash__(self):
             # `.md` excluded — doc-only edits do not affect the Gradle build.
             r"^java-sdk/(?!.*\.md$).*",
         ],
+        FileGroupForCi.JAVA_SDK_CONFORMANCE_FILES: [
+            # The Java SDK, plus what its serialization conformance check compares it against:
+            # Airflow's serializer and schema, and the shared harness.
+            r"^java-sdk/(?!.*\.md$).*",
+            r"^airflow-core/src/airflow/serialization/serialized_objects\.py$",
+            r"^airflow-core/src/airflow/serialization/schema\.json$",
+            r"^scripts/ci/lang_sdk_serialization/.*",
+        ],
         FileGroupForCi.TS_SDK_DOCS_FILES: [
             # TypeDoc renders the reference from the SDK sources and category entry points,
             # and the landing page is authored in ts-sdk/docs — unlike TS_SDK_FILES, `.md`
@@ -1942,6 +1951,11 @@ def skip_prek_hooks(self) -> str:
             # from Maven Central. Skip it when no java-sdk files changed so unrelated PRs do not
             # depend on that resolution.
             prek_hooks_to_skip.add("regenerate-java-sdk-verification-metadata")
+        if not self._matching_files(FileGroupForCi.JAVA_SDK_CONFORMANCE_FILES, CI_FILE_GROUP_MATCHES):
+            # This hook compiles the Java SDK with Gradle and resolves its dependencies from Maven
+            # Central. Skip it unless a java-sdk file, Airflow's serializer or schema, or the shared
+            # harness changed. Those last do not force full_tests_needed, so they need their own group.
+            prek_hooks_to_skip.add("check-java-sdk-serialization-conformance")
         if not self._matching_files(FileGroupForCi.TS_SDK_FILES, CI_FILE_GROUP_MATCHES):
             # This hook regenerates ts-sdk/src/generated/supervisor.ts from the wire schema and
             # diffs it. Schema-only changes deliberately do not trigger it: regenerating the