Java SDK: Serialize native Dags to DagSerialization v3 on parse requests

A Java-authored Dag could not reach the scheduler on its own: the runtime
answered task-execution requests only, so a Python stub file still had to
exist purely to describe the Dag's structure. With dependency edges and
schema-keyed configuration now recorded on the Java Dag model, the runtime
has everything it needs to answer the coordinator's DagFileParseRequest the
same way the Go SDK does, and the nativedag examples become real schedulable
Dags with no Python counterpart.

Native Java tasks deliberately emit no _arg_bindings: the execution API
delivers bindings only for Python _StubOperator tasks, and a Java task
always runs inside the JVM bundle that already holds its wired inputs, so
the runtime resolves them locally.

Cron schedules map to CronTriggerTimetable only, mirroring the Go SDK until
the supervisor forwards the [scheduler] timetable flags over the coordinator
protocol (see the TODO at the timetable serializer).

(cherry picked from commit 651791a)

Author: jason810496

java-sdk/README.md

@@ -619,17 +619,17 @@ prek hook regenerate it.
 | 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 | – |  |
+| capability: `native-dag-authoring` | SHOULD | ✓ | 3.3 |  |
+| capability: `task-args` | MUST † | ✓ | 3.3 |  |
+| capability: `dag-params` | MUST † | ✗ | – |  |
+| capability: `taskflow-dependencies` | MUST † | ✓ | 3.3 |  |
+| capability: `branching` | SHOULD † | ✗ | – |  |
+| capability: `dag-test` | SHOULD † | ✗ | – |  |
+| capability: `task-group` | MAY † | ✗ | – |  |
+| capability: `dynamic-task-mapping` | MAY † | ✗ | – |  |
+| capability: `asset-inlets-outlets` | MAY † | ✗ | – |  |
+| capability: `asset-scheduling` | MAY † | ✗ | – |  |
+| capability: `object-store` | MAY † | ✗ | – |  |
 
 *Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.*
 

java-sdk/capabilities.yaml

@@ -52,8 +52,8 @@ states:
     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.
+# Runtime capabilities reflect the task-facing Client surface. Native-Dag capabilities describe
+# what a Java-authored Dag can declare and deliver through Dag serialization.
 capabilities:
   mixed-lang-stub-target:
     supported: true
@@ -95,14 +95,16 @@ capabilities:
     supported: false
     note: "no task-facing asset-event API yet"
   native-dag-authoring:
-    supported: false
-    note: "native Dag authoring not implemented yet"
+    supported: true
+    since: "3.3"
   task-args:
-    supported: false
+    supported: true
+    since: "3.3"
   dag-params:
     supported: false
   taskflow-dependencies:
-    supported: false
+    supported: true
+    since: "3.3"
   branching:
     supported: false
   dag-test:

java-sdk/sdk/module.md

@@ -56,17 +56,17 @@ meaning of each dimension is defined in the
 | 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 | – |  |
+| capability: `native-dag-authoring` | SHOULD | ✓ | 3.3 |  |
+| capability: `task-args` | MUST † | ✓ | 3.3 |  |
+| capability: `dag-params` | MUST † | ✗ | – |  |
+| capability: `taskflow-dependencies` | MUST † | ✓ | 3.3 |  |
+| capability: `branching` | SHOULD † | ✗ | – |  |
+| capability: `dag-test` | SHOULD † | ✗ | – |  |
+| capability: `task-group` | MAY † | ✗ | – |  |
+| capability: `dynamic-task-mapping` | MAY † | ✗ | – |  |
+| capability: `asset-inlets-outlets` | MAY † | ✗ | – |  |
+| capability: `asset-scheduling` | MAY † | ✗ | – |  |
+| capability: `object-store` | MAY † | ✗ | – |  |
 
 *Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.*
 

java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt

@@ -33,8 +33,10 @@ import kotlinx.coroutines.runBlocking
 import org.apache.airflow.sdk.execution.CoordinatorComm
 import org.apache.airflow.sdk.execution.LogSender
 import org.apache.airflow.sdk.execution.Logger
+import org.apache.airflow.sdk.execution.comm.DagFileParseRequest
 import org.apache.airflow.sdk.execution.comm.ErrorResponse
 import org.apache.airflow.sdk.execution.comm.StartupDetails
+import org.apache.airflow.sdk.execution.parseDags
 import org.apache.airflow.sdk.execution.runTask
 import kotlin.text.substringAfterLast
 import kotlin.text.substringBeforeLast
@@ -181,6 +183,7 @@ class Server(
     val frame = coordinator.readMessage()
     when (val body = frame.body) {
       is StartupDetails -> runTaskAndReport(bundle, body, coordinator)
+      is DagFileParseRequest -> parseDagsAndReport(bundle, body, coordinator)
       is ErrorResponse -> throw ApiError("[${body.error}] ${body.detail}")
       else -> throw ApiError("Unexpected initial frame (id=${frame.id})")
     }
@@ -194,4 +197,12 @@ class Server(
     val result = runTask(bundle, startup, coordinator)
     coordinator.communicate<Unit>(result)
   }
+
+  private suspend fun parseDagsAndReport(
+    bundle: Bundle,
+    request: DagFileParseRequest,
+    coordinator: CoordinatorComm,
+  ) {
+    coordinator.communicate<Unit>(parseDags(bundle, request))
+  }
 }