Java SDK: Serialize native Dags to DagSerialization v3 - #71190
Draft
jason810496 wants to merge 10 commits into
Draft
Java SDK: Serialize native Dags to DagSerialization v3#71190jason810496 wants to merge 10 commits into
jason810496 wants to merge 10 commits into
Conversation
The @task.stub TaskFlow support in providers-standard imports KNOWN_CONTEXT_KEYS, PlainXComArg, MappedOperator and the decorator base classes through the compat layer so the provider keeps working down to Airflow 2.11. Those symbols first ship in common-compat 1.19.0 (1.18.0 was released from main in the meantime without them), so the version is cut here for the standard provider's pin to resolve.
Stub tasks silently ignored TaskFlow call arguments, so a Dag author could not hand literals or upstream XCom results to a lang-SDK runtime. The decorator now binds the call to the stub's signature at parse time and captures an ordered arg spec (literal values and direct upstream XCom references, with pydantic-derived JSON value schemas) that serializes with the Dag, while rejecting what cannot cross the language boundary: custom XCom keys, aggregated mapped outputs, non-JSON literals, and stubs with arguments inside mapped task groups. Mapped (.expand()) stubs capture no spec and keep the legacy behavior until a follow-up delivers per-map-index bindings.
TIRunContext gains an arg_bindings field so a lang-SDK runtime receives the stub task's TaskFlow arg spec at startup. ti_run derives it from the serialized Dag only for stub operators, so regular tasks never pay for the lookup, and only for clients on the new API version -- gated on the Cadwyn VersionChangeWithSideEffects.is_applied check rather than a date comparison -- so stub Dags that predate arg bindings keep running against older clients, for which the version migration strips the field.
StartupDetails in the supervisor wire schema carries the new arg_bindings so foreign runtimes receive the spec at task startup, with a version migration that strips it for runtimes pinned to the previous schema. The Go and TS SDKs regenerate against the new schema version; the Go arg-binding runtime itself lands in a stacked follow-up PR.
An XComArg buried in a list or dict literal fell through to the JSON check, whose "pass it in its JSON form instead" advice is impossible to follow for a task output. Detect nested references up front and point the author at the working alternative: pass the upstream output as its own argument.
When a PR cuts a new provider version while the previous version is still being voted on, only the rcN tags exist on the apache remote - the final tag is pushed after the vote passes. The changes-table walk in _get_all_changes_for_package assumed every past version has a final tag and crashed with git exit 128 in that window, breaking CI for any PR that bumps a provider version during a release wave.
`dag.addTask("extract", Extract.class)` stored tasks as a plain
`Map<String, Class<out Task>>`, which leaves nowhere to hang anything
else a task needs: dependency edges, task-level configuration, and
argument wiring all have to attach to a per-task object, and a map of
classes cannot carry them. Introducing that object now keeps those
follow-ups additive instead of forcing another break of the registration
API later.
The annotation surface keeps `Builder.Dag` / `Builder.Task`, and the
interface users implement keeps the `Task` name, so the definition
objects are `DagDef` and `TaskDef` -- a pairing that stays unambiguous
next to `Task` at a use site. The SDK is pre-1.0, so the old
string-keyed overload is removed outright rather than deprecated.
For a stub-backed Dag the Python file's `@task.stub` call site is the graph the scheduler actually orders the run by, so it must also be what feeds the Java task its inputs. The Java side previously re-declared that data flow with `@Builder.XCom(task = "...")`, duplicating the Dag file's wiring in a second place that nothing keeps honest: rename or re-wire a task in Python and the Java annotation silently keeps pulling the old upstream. The 2026-10-30 supervisor schema delivers the call site's bindings with every task run, so the runtime can read them instead of guessing. Binding is positional, matching the Go SDK's flat-parameter contract: Java parameter names are not API, so an IDE rename must never rebind an input. Keyword-style calls bind by name only through an explicit `TaskInput` bundle whose public fields declare their wire names -- the deliberate, tagged boundary for snake_case-to-camelCase crossings. A task declares flat data parameters or one bundle, never both, so field names and positions cannot shift each other. jsonSchema2Pojo cannot express the kind-discriminated binding union, so the generated `TIRunContext` carries the raw payload and a small hand-written decoder materializes the typed view.
Until now the Java SDK could only supply task bodies: a Python @task.stub Dag had to own the schedule, every task option, and the graph. That splits one pipeline across two languages and two repositories for no reason other than a missing authoring surface, and it left the Java-side model with nothing to describe -- no edges, no configuration -- so there was nothing a native Java Dag could be built from. Java annotations cannot change call semantics the way Python decorators do, so the graph is declared against a compile-time-generated twin class (`<Class>Ref`): calling a twin registers the task and passing one twin's handle into another feeds the upstream's output into the downstream's parameter, making the call graph the task graph the way Python TaskFlow does -- but type-checked by javac through the In/TaskRef generics. Keeping the wiring calls the only way to express an edge means there is one graph story to learn instead of two, and the method stays optional so stub-backed classes are unchanged: their graph still lives in the Python Dag file, and runtime arg bindings continue to win over anything Java declares, because for a stub task the Python call site is the graph the scheduler ordered the run by. Dag and task configuration is generated from Airflow's own Dag serialization schema rather than hand-listed, so the Java attributes cannot drift from the Python semantics they mirror and new scalar keys appear after a schema sync. Only attributes written at the use site are applied, leaving Airflow's defaults in charge of everything unset. Design rationale is recorded in ADR-0007.
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).
boring-cyborg
Bot
added
area:API
This was referenced Aug 5, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Why
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 — even for a Dag whose schedule, configuration and dependency edges were all declared in Java. With edges and schema-keyed configuration now recorded on the Java Dag model, the runtime has everything it needs to answer the coordinator's
DagFileParseRequestitself, and thenativedagexamples become real schedulable Dags with no Python counterpart.How
parseDagsserializes every Dag registered on theBundleto DagSerialization v3 — the same shape Python'sDagSerializationemits — and returns it as aDagFileParsingResultbody. It mirrors the Go SDK's serde (go-sdk/pkg/execution/serde.go):type/varenvelopes, omit-if-schema-default operator fields, sorted tags and downstream ids, and a timetable derived from the schedule.Server.dispatchTaskgains aDagFileParseRequestbranch alongsideStartupDetails, so a bundle process serves either a task run or a parse request.[core]config rather than a JSON-schema default (max_active_tasks_per_dag,max_active_runs_per_dag) are always emitted, falling back to the same values._arg_bindings: the execution API delivers bindings only for Python_StubOperatortasks, and a Java task always runs inside the JVM bundle that already holds its wired inputs, so the runtime resolves them locally.What
execution/Serde.kt(parseDags,serializeDag,serializeTask, timetable/task-group/value serializers) withSerdeTestcovering required fields, config emit rules, wiring-recorded edges, theDagFileParsingResultenvelope, and temporal/nested-map encoding.Server.dispatchTask, covered by a newServerTestcase that drives a realDagFileParseRequestframe end to end.Known limitation
Cron schedules map to
CronTriggerTimetableonly, mirroring the Go SDK until the supervisor forwards the[scheduler]timetable flags over the coordinator protocol. There is aTODOat the timetable serializer.Was generative AI tooling used to co-author this PR?