Java SDK: Honor TaskFlow arg bindings sent by the supervisor - #71188
Draft
jason810496 wants to merge 8 commits into
Draft
Java SDK: Honor TaskFlow arg bindings sent by the supervisor#71188jason810496 wants to merge 8 commits into
jason810496 wants to merge 8 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.
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
For a stub-backed Dag, the Python file's
@task.stubcall site is the graph the scheduler actually orders the run by — so it should also be what feeds each Java task its inputs. Until now the Java side re-declared that data flow itself, with@Builder.XCom(task = "extract")naming the upstream to pull. That is the same wiring written twice, in two languages, with nothing keeping the copies honest: rename or re-wire a task in the Dag file and the Java annotation keeps pulling the old upstream, silently. It also cannot express a literal written at the call site at all.The 2026-10-30 supervisor schema delivers the call site's argument bindings with every task run, so the Java runtime can read the real wiring instead of restating a guess at it.
How
ClientandContext— resolve the binding at their index, in declaration order. This matches the Go SDK's flat-parameter contract, and it is deliberate: Java parameter names are not API, so an IDE rename must never rebind an input.TaskInputbundle parameter, whose public fields declare their wire names (@ArgName("region_code"), or the verbatim field name). That is the one tagged place where the stub'ssnake_caseargument names cross intocamelCaseJava fields, rather than an implicit convention applied everywhere.null, so it fails withMissingXComExceptionwhen its binding resolves to nothing; boxed and reference types receivenull. Declaring more data parameters than the call site bound fails the task rather than running it with missing inputs.Client.hasArgs/hasArg/getArg, by position or by name.@Builder.XComis removed outright rather than deprecated — the SDK is pre-1.0, and leaving a second, diverging way to declare the same data flow is the problem this change fixes.kind-discriminated binding union, so the generatedTIRunContextcarries the raw payload and a small hand-written decoder materializes the typed view.What
2026-10-30and refresh the schema snapshot, which addsTIRunContext.arg_bindings.ArgBinding+decodeArgBindings,TaskInput,@ArgName,internal.ArgValues, andClient.hasArgs/hasArg/getArg.BuilderProcessorclassifies each@Builder.Taskparameter asClient,Context, aTaskInputbundle, or a flat data parameter, and emits the matchingArgValuesresolution;@Builder.XComand its codegen are gone.TaskInputbundle bound from keyword arguments, and document argument binding in the Java SDK docs.Was generative AI tooling used to co-author this PR?