Java SDK: Author complete Dags in Java without a Python stub file - #71189
Draft
jason810496 wants to merge 9 commits into
Draft
Java SDK: Author complete Dags in Java without a Python stub file#71189jason810496 wants to merge 9 commits into
jason810496 wants to merge 9 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.
boring-cyborg
Bot
added
area:API
This was referenced Aug 5, 2026
phanikumv
reviewed
Aug 5, 2026
phanikumv
left a comment
phanikumv
left a comment
Contributor
There was a problem hiding this comment.
this PR seems way too huge, is there a way to create smaller PRs so that it is easier to review?
Member
Author
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
2 participants
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
Until now the Java SDK could only supply task bodies. A Python
@task.stubDag had to own the schedule, every task option, and the graph — so a pipeline whose logic is entirely Java still had to be described in two languages, in two places, for no reason other than a missing authoring surface. It also left the Java-side model with nothing to describe:DagDefheld anid -> task classmap, no edges and no configuration, so there was nothing a native Java Dag could be built from and nothing to serialize for one.Python's TaskFlow shows the shape worth matching: calling tasks like functions is the graph declaration —
load(transform(extract())). Java annotations cannot change call semantics the way Python decorators do (invoking the real method would run its body), so the call syntax has to target something generated at compile time.How
@Wiringmethod receives a generated<Class>Refclass whose methods mirror the@Builder.Taskmethods: injectedClient/Contextparameters are dropped, data parameters takeIn<T>, and the return value is aTaskRef<T>. Calling a twin registers the task; passing one twin's handle into another feeds the upstream's output into the downstream's parameter and records the edge. The call graph is the task graph, andjavacchecks it — numeric parameters accept any numeric upstream (In<? extends Number>),Object/rawMap/rawListaccept any (In<?>), everything else accepts covariant matches. Unknown upstreams are unrepresentable and cycles are unconstructible in call syntax.@Builder.Taskmethod the wiring never invoked fails at Dag-parse time.@Wiringis optional, so stub-backed classes are untouched. A class without one registers every task with no Java-side edges — exactly today's behaviour. The existing stub-backed examples gain neither@Wiringnor configuration.arg_bindings, the binding at a parameter's position is what the task receives; for a stub task the Python call site is the graph the scheduler ordered the run by, so the Java class must not be able to disagree with it. Wired inputs are the fallback — the native-Dag case, where no Python call site exists. Binding is positional either way: Java parameter names are not API, so an IDE rename must not rebind an input.TaskSpecgenerator (scalars only, serializer-owned keys skipped, a documented exclusion list that fails generation when it goes stale, a hand-curated Dag-level allowlist). Only attributes written at the use site are lowered intoconfigcalls, so Airflow's own defaults still govern everything left out.Builderclass is generated, outer class and both nested annotations, so there is exactly one definition of it;id(andtoonDag) stay the leading structural attributes, and generation fails if a schema key ever camel-cases onto one of them.Bundleconstruction validates what the type system cannot.TaskDef.dependsOncan express a cycle or point at a task in another Dag, so the bundle checks acyclicity and same-Dag upstream membership at parse time.What
@Wiring,In/TaskRef,internal.Refs(twin registration), andinternal.Fields(config validation).Builder(with schema-derived@Builder.Dag/@Builder.Taskconfiguration attributes) andinternal.SchemaFieldsfrom a vendoredsdk/schema/dag-schema.json, kept in sync withairflow-coreby the newsync-java-sdk-dag-schemaprek hook; delete the hand-writtenBuilder.kt.BuilderProcessoremits the<Class>Reftwin, aDAG_IDconstant and adag()factory, lowers explicit annotation attributes intoconfigcalls (validating ISO-8601 temporals at compile time), and verifies the wiring registered every task.DagDef.config,TaskDef.config/dependsOn/inputs, anaddTask(task, upstreams)overload, cycle and upstream validation inBundle, andContext.taskDefthreaded through the task runner.ArgValuesfalls back to the@Wiring-recorded inputs when the supervisor sent no bindings, and exposeshasRuntimeBindingsso aTaskInputbundle is filled field-by-field from bindings but decoded wholesale from its single wired input otherwise.nativedag/examples in both styles (registered in the example bundle), the ADR atairflow-core/adr/lang-sdk/0007-taskflow-dag-dsl.md, and a "Native Java Dags" section in the Java SDK docs.Was generative AI tooling used to co-author this PR?