* Java SDK: Read the Dag run conf into a class with typed fields
Signed-off-by: PoAn Yang <payang@apache.org>
* Document the clashing field names case of getConf
Signed-off-by: PoAn Yang <payang@apache.org>
* Reject nulls for primitives in getConf and document JsonProperty
Signed-off-by: PoAn Yang <payang@apache.org>
* Move getConf details from the Java SDK guide to the API reference
Signed-off-by: PoAn Yang <payang@apache.org>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
These checks build and run each SDK's serializer, so as prek hooks they needed a full
Go, Java or Node toolchain and took minutes on every commit touching Airflow's
serializer. They are tests, not static checks, so they now run as a CI test job that
sets up only the toolchain of the SDK it checks, for the SDKs whose inputs changed.
* Load the serialized Dag in ti_run only for stub tasks
Starting any task deserialized its whole Dag so the route could find out
whether the task was a stub and read its arg_bindings. Denormalize the
serialized is_stub flag onto task_instance, which ti_run already fetches
and locks, and load the Dag only when it is set.
execution_timeout is resolved from the same definition, so it is now sent
for a lang-SDK task only; #75387 has a Python task send its own to the
supervisor.
* Hold SQLite foreign keys off while the downgrade rebuilds
Dropping a column rebuilds the table on SQLite, unlike adding one, so the
downgrade copies task_instance rows while other tables still point at it.
* Pin the bulk-insert path and reword the is_stub comments
create_dagrun refreshes every row from its task right after the bulk write,
so the existing test passed even when insert_mapping dropped the key. Assert
the mapping itself instead, and record what a row keeps when it moves to a
new Dag version without being refreshed.
* Mark a Go or Java TriggerDagRun task a stub, as TypeScript does
* Inline the table name in the is_stub migration
---------
* Add consolidated Language SDK compatibility matrix
Users comparing Language SDKs had to read each SDK's README, and the
per-SDK manifests lagged behind what the SDKs actually ship. One generated
matrix, split by authoring mode, shows what each SDK supports for a Dag
definition and for a Python Dag with stub TaskHandler.
* Add callbacks, listeners and provider operators to the SDK matrix
Users comparing Language SDKs could not see where callbacks, Airflow's
listener API and Python provider operators stand, because the matrix had
no dimension for any of them. Callbacks ask different things of an SDK per
authoring mode, so they are declared per mode like the other shared
capabilities.
* Address review on the Language SDK matrix generator
The umbrella capability and the authoring mode both read `dag-definition`,
which gives one name two meanings in the same manifest. A row's kind also
belongs in its entry key: state and capability names happen not to collide
today, so the lookup worked by accident rather than by design.
* Rename the dynamic-task-mapping dimension to mapped-tasks
The dimension is about whether a task can be a mapped task, which the
shorter name says directly and keeps the matrix row readable next to the
other task-level dimensions.
* Split callbacks into Dag-level and task-level dimensions
The callbacks of a Python Dag run whichever SDK executes its stub tasks, so
``dag-callbacks`` is a Dag-definition concern only, while ``task-callbacks``
stays declared per authoring mode. Drop ``provider-operators`` until ADR-0009
is accepted, and reuse the shared ``capabilities_doc`` fixture in the per-SDK
README tests.
* Fix the hook attribution in the matrix generation diagram
``index.rst`` is written by ``update-lang-sdk-central-matrix`` from every SDK
in ``LANG_SDKS``, not by the per-SDK README hook.
---------
* Java SDK: Rewrite the user docs around Dag definition
Reorder the Java SDK guide to start from a Dag defined in Java, document
conditions, switches and TriggerDagRun, and add a table of how stub task
arguments bind. Split the README into a user landing page and
DEVELOPMENT.md, and add an example README.
* Add Logback and appender to the docs spelling wordlist
* Address review: Dag-level queue, version sync, dropped constraints
Set the queue once on the Dag in every example, now that a Java Dag takes
one, and bump the documented version to 1.0.0-beta2 everywhere. Add a prek
hook that fails when the guide, the README and the Dokka landing page name
different versions. Restore the TaskInput constraints, correct the
LocalExecutor wording, and drop the Python Label comparison.
* Generate the quick start of the Java SDK README and Dokka page
The quick start appears on three pages. The guide owns it now, between
SHARED SNIPPET markers, and the sync-java-sdk-docs hook copies it into the
other two and still checks that all three install the same version.
* Address review: install a version placeholder, not a release
The three pages that show how to install the SDK pinned a release that is
not out yet, which also held the PR back. They now install `<version>` and
link to Maven Central, and the sync hook fails when a page names a release
instead. The parse-failure bullet leads with the SDK release a JAR was built
with, and keeps the supervisor schema version as the explanation.
---------
This reverts commit eda782f76e.
Since it merged, the K8S Lang-SDK test has hung in about 40% of CI runs:
a Java task stops right after its first request to the supervisor and
stays running until the test times out. Every run before the merge
passed. Reverting keeps CI usable while the hang is investigated on
https://github.com/apache/airflow/pull/74690.
Generated-by: Claude Opus 5
* Java SDK: Let a native Java Dag set one queue for all its tasks
Every task of a native Java Dag runs on the Java coordinator, so each one
needs a queue that queue_to_coordinator routes there. Airflow's Dag has no
queue setting, so authors had to repeat the same queue on every task. The
Go and TypeScript SDKs already take a Dag-level queue that each task
inherits unless it sets its own; this gives Java the same.
* Reject a blank Dag-level queue and fix the native Dag queue docs
A blank queue matched no coordinator, so every task stayed queued.
The parsing docs still said there was no Dag-level queue and routed a
queue the examples do not use.
* Reject a blank Dag-level queue in the annotation processor
* Carry the blank rule on the schema field, for Dag and task queues alike
* Set the queue once per Dag in the e2e native bundle
* Put the blank check on Field and tidy the comments
---------
* Make Java SDK context signal-aware
Signed-off-by: PoAn Yang <payang@apache.org>
* Share the stop request with every context made during a task
Signed-off-by: PoAn Yang <payang@apache.org>
* Update java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Context.kt
---------
Signed-off-by: PoAn Yang <payang@apache.org>
Co-authored-by: Jason(Zhe-You) Liu <68415893+jason810496@users.noreply.github.com>
* Java SDK: Send log timestamps in UTC
Signed-off-by: PoAn Yang <payang@apache.org>
* Fix Java SDK log time zone and a supervisor crash on bad timestamps
Signed-off-by: PoAn Yang <payang@apache.org>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
* Java SDK: Add edge labels to native Dags
Signed-off-by: PoAn Yang <payang@apache.org>
* Apply edge labels in the order they were set
A task edge's own label always won over a group edge's label, and a
relabeled group edge kept its first position, so edge_info differed from
Python's, where the label set last wins. Also note that Python's Label
rewrites an edge when one end is in no task group.
* Java SDK: Write the edge label text before the task
Signed-off-by: PoAn Yang <payang@apache.org>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
Co-authored-by: ZHE YOU LIU <zhu424.dev@gmail.com>
Hard SDK validation can go stale when the server changes the allowed ID
rules, and the server already validates authoritatively, so the SDK only
warns. Per review, the check runs at the interface based syntax, which
both authoring styles compile down to: a checkAirflowBundle verification
task in the check lifecycle runs BundleInspector (the ADR-0003 shape)
against the assembled Bundle and prints a warning for every ID the
server would reject. Warnings never fail the build; a bundle that cannot
be built at all does.
Mapped tasks were the one kind of dynamic work still outside the region
model: their instances sat in the sentinel region and were told apart only by
map_index. A loop has to be able to contain an expansion, one per pass, and that
only works if the expansion owns a region of its own.
The coordinate column is renamed in place so large deployments pay for metadata
changes, not a table rewrite. No table is rebuilt because SQLite's rebuild would
cascade-delete the rows that reference task_instance.id. legacy_task_data_owner,
xcom_v1 and rtif_v1 keep map_index because they record the coordinates their
rows had before regions existed. The ORM keeps map_index as an alias: Jinja
log-filename templates and provider code still read ti.map_index from ORM rows,
and removing it needs its own audit.
Mapping keeps its existing behaviour. A task's region is born with its
placeholder, re-expansion grows or shrinks it in place, and a map never forks.
Promoting the placeholder to index 0 stays in place too. Attempts retired before
the promotion stay at index -1: they own their XCom, rendered fields and log
files by UUID, and moving them could collide with retired attempts already
at index 0. A pre-region placeholder never ran, so it has no history to keep;
it moves into a freshly minted region and is promoted like any other, instead
of being retired as a phantom failed try. Pre-region expansions stay in the
sentinel region and keep revising in place, so the upgrade does not rewrite
existing rows.
Callers that address a task instance by public coordinates now find a task's own
region as well as the sentinel. The DagRun wait endpoint reads current producers
only and rejects loop results, because which pass's result to return would be
ambiguous.
A task in a loop or mapped region is identified by (region_id, region_index),
but running code, the Execution API and log readers still address it by the
public map_index. If one value served both, a loop pass would look like a mapped
expansion. We derive the public index from the Dag definition pinned to the try
instead of storing it: a task exposes its region index as map_index only when
its pinned definition expands, and -1 otherwise. Ordinary mapped tasks therefore
keep reporting exactly what they did before.
The scheduler and API server must understand loops without importing user code,
so the serialized Dag carries a loop group definition they can read.
The Execution API and the supervisor schema take the region fields as optional,
versioned fields so that older task SDKs keep working against a newer API
server. Lookups of earlier runs refuse loop producers, because an iteration of
one run has no counterpart to resolve in another run.
A log path and a remote log id have to come out the same for the process that
writes the log and for every reader that fetches it later, retired attempts
included, and one loop pass must not overwrite the logs of the pass before it.
Building both from one prepared, batched context gives workers, the triggerer
and log readers the same answer without a query per task. Only loop tasks gain
a pass position; ordinary mapped tasks keep their existing path, so logs already
written stay readable.
* Java SDK: Trigger another Dag's run from a native Java Dag
A native Java Dag could not start another Dag's run. Airflow's
TriggerDagRunOperator is Python, and a Java bundle has no Python Dag file to
declare one in, so a Java Dag that fans out to another Dag needed a Python
counterpart for that one task.
A `TriggerDagRun` passed to `dag.task` declares a task that has no Java body
to run; the Java runtime runs it as `TriggerDagRunOperator` does. Its
settings are named as Python names the operator's parameters:
```java
dag.task("trigger_downstream",
new TriggerDagRun("downstream_etl")
.config("wait_for_completion", true)
.config("conf", Map.of("rows", 2)));
```
An annotated Dag declares one from an ordinary `@Builder.Task` method that
returns a `TriggerDagRun`. The serialized Dag has to carry what is
triggered, so that method runs when the Dag is built rather than when the
task runs, and therefore takes no parameters.
The task serializes as Python's own `TriggerDagRunOperator`: its template
fields, its UI colour and extra link, and a `dag_dependencies` entry, so the
Airflow UI draws the dependency between the two Dags. Python writes only an
operator's template fields and keeps the rest in the Dag file; a Java Dag
has no such file, so every setting the author made is written too.
The runtime records the triggered run's link and ID as XComs, skips or fails
on a run that already exists, and waits for completion either by polling or,
with `deferrable`, by deferring to the Python triggerer's DagStateTrigger and
resuming from the event it fires. That makes this the first Java runtime to
emit DeferTask and a skipped TaskState.
`scripts/ci/lang_sdk_serialization/test_dags.yaml` now gives the shared
trigger Dag a real logical date instead of a Jinja template, because the SDK
takes one as a date-time. The Go SDK is making the same change in #74417;
whichever lands second drops that one line, and until then the Go
serialization conformance check reports that single difference.
* Java SDK: Point both example triggers at a Dag that triggers nothing
* Java SDK: Say why a switch cannot choose a task that triggers a Dag run
* Java SDK: Snapshot a TriggerDagRun when its task is declared
* Java SDK: Check a TriggerDagRun setting the way every other config key is checked
* Java SDK: Describe what a trigger task pushes and how it names a run
* Java SDK: Correct which runtimes already run trigger tasks
* Java SDK: Format a trigger task's date-times in one place
* Java SDK: Reject a trigger method that throws a checked exception
* Java SDK: Give the trigger fixture a logical date every SDK can write
* Java SDK: Run a trigger task through runTask in tests
* Java SDK: Check a trigger task polls the run it started
* Java SDK: Choose one of several tasks with a branch
A condition picks between two tasks, but a Dag that routes work by size or
by region needs more than two, and splitting that into nested conditions
makes the graph unreadable.
`DagDef.Branch` and `@Builder.Branch` declare a task that chooses one of
the tasks listed with `option(...)`; every other one is skipped. The chosen
task's ID becomes the branch's return value, as it does for Python's branch
operators. A branch chooses exactly one case, so a task that runs after
several of them needs a trigger rule such as
`none_failed_min_one_success`.
Each authoring style names the case in the way that style can type-check. A
`BranchTask` returns the case's own class, so javac checks it; two cases
that share a class are rejected where the second is named, since the branch
could not tell them apart. A `@Builder.Branch` method has no such class to
name -- a task method's implementation is generated -- so it returns a
`TaskId` from the generated `<Dag>Builder.TaskIds`, which holds one
constant per task of the Dag. Either way nothing hard-codes a task id
string, and the generated holder is emitted only for a Dag that declares a
branch.
* Java SDK: Name the multi-way branch Switch, as the other SDKs do
* Java SDK: Name a switch case by its task class in both styles
* Java SDK: Say when a switch rejects a choice and when it fails
* Java SDK: Cover a generated switch that names a task it cannot choose
* Java SDK: Check a switch carries the chosen task's id
* Java SDK: Let a wiring view name a switch's cases in more than one call
* Java SDK: Pick one of two tasks with a condition
A native Java Dag could only run every task it declared: it had no way to
let a task's result decide what runs next, so a Dag that loads only when
there are rows had to be split in two or fall back to a Python stub.
`DagDef.If` and `@Builder.If` declare a task whose boolean names the side
that runs; the other is skipped. The deciding task carries
`_can_skip_downstream`, so Airflow skips a cleared side again rather than
running it, and the SDK writes the `skipmixin_key` XCom before the skip, as
Python's SkipMixin does. It skips only the side not taken, so a task after
both sides needs a trigger rule such as `none_failed_min_one_success`.
`If` is capitalized because `if` is a Java keyword and `@Builder.If` reads
the same way. With no task id given, one is derived from the condition
class's simple name, so `HasRows.class` becomes `hasRows`.
The runtime reaches `SkipDownstreamTasks` through the transport client
only, never through the task-facing `Client`, so a task body cannot skip
tasks the Dag did not declare it may.
* Java SDK: Let a wiring view name a condition's sides in more than one call
* Java SDK: Reject a decider class registered as an ordinary task
* Java SDK: Name a condition's sides Then and Else
* Java SDK: Pin the order a condition skips and records what it skipped
* Java SDK: Drop the unused switch in the condition test's decider
* Java SDK: Record the class that declares each Dag and describe it on request
DagDef remembers the outermost class that built it, and a generated
@Builder.Dag builder names the annotated class instead of itself.
Server accepts --describe-sources <file> to write each Java-declared Dag's
declaring class and return without connecting, for the Gradle plugin.
* Java SDK: Pack each native Dag's source file into the bundle JAR
The packDagSources task runs mainClass in describe mode, finds each Dag's
source file through the class file's SourceFile attribute, and packs each
unique file with a sources.json mapping under META-INF/airflow/. The JAR
manifest names it with Airflow-Java-SDK-Sources.
* Java SDK: Bound the describe run and keep its payload to the bundle JAR
Every `Jar` task gained a dependency on `packDagSources`, so a sources or
javadoc JAR carried the Dag sources and claimed the
`Airflow-Java-SDK-Sources` attribute that the Python side picks a bundle
JAR out of a directory by. Only `jar` and `shadowJar` carry them now.
The describe run had no time bound, so a `main` that does not route
through `Server.create(args)` and starts serving instead held the build
open with no way out. It now runs under a timeout and falls back to
packing the entrypoint's source alone, which is what every other
describe failure already does.
Say in `serve` and `serveAsync` that a `--describe-sources` server
returns without connecting, and record in `sources.json` that its paths
are relative to the source directory, which is also where they sit under
`META-INF/airflow/sources/`.
* Java SDK: Show each Dag's own source and find it outside the package layout
The Java importer now passes the Dag ID to the source lookup, so the Code
view reads dag_source_paths instead of always showing the entrypoint.
The source locator falls back to a unique match on the SourceFile name
when the package does not match the directory, using an index built only
when the package path misses. An unresolvable entrypoint or Dag class is
now logged as a warning.
The describe task is no longer up to date after a failed run, so a
transient failure does not stick until a source changes. Its hint now
points at Server.create(args), which is what answers --describe-sources.
Add a shadowJar test, correct the comment on which JARs carry the sources
payload, and document that a Dag maps to the class that constructed it.
* Java SDK: Use a describe run's output even when the run itself failed
* Java SDK: Serve and describe sources from their own Server classes
* Java SDK: Pack the sources into the one JAR the bundle is built from
* Java SDK: Attach the plugin's documentation to the plugin class
* Java SDK: Drop the duplicated classes dir from the describe classpath
* Java SDK: Let a bundle set how long the describe run may take
* 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).
* Test that the conformance check compares Dag params
* Java SDK: Leave config-backed Dag fields for Airflow to fill in
* Java SDK: Leave out the email flags a Java task cannot act on
* Java SDK: Serialize a native Dag's task groups
* Java SDK: Write each Dag tag once, as Python does
* Java SDK: Check native Dag serialization against Airflow's
* Java SDK: Run the serialization conformance check on every file that steers it
`Bundle.register` expands the group edges, `Refs` decides which group each
task lands in, and `sdk/build.gradle.kts` generates the schema fields the
serializer writes from. All three change the serialized output, so add
them to the hook's trigger set.
Report a failed classpath build with the Gradle output instead of an
`IndexError` on an empty line list or a bare `CalledProcessError`.
* Java SDK: Build a native Dag's timetable the way Python does
Expand a cron preset before writing it, so `@daily` serializes as
`0 0 * * *` and the Dag hashes the same as the Python one. Reject a
schedule that is neither a preset nor a cron expression where the Dag is
written, rather than letting it reach a scheduler that cannot build a
timetable from it.
Take the timezone from the Dag's start date, as Python's `DAG.timezone`
does, instead of always writing UTC: a Dag started at a non-UTC offset
now fires at that offset rather than at the same wall clock in UTC.
* Java SDK: Carry a native Dag's call arguments as argument bindings
A native Dag's serialized form recorded its task edges but nothing about
what each task was called with, so changing a literal argument left the
Dag byte-identical: no new version, and nothing in the version diff.
Write the same `is_stub` flag and `_arg_bindings` spec the TypeScript SDK
writes, as ADR-0007 decision G says a natively authored Dag should. The
wiring view now passes its parameter names along with the arguments, so
each binding can name the parameter it feeds.
A Java task still resolves its arguments from the Dag in its own bundle
rather than reading the spec back: the bundle that runs the task also
holds the call that wired it, so the values never have to travel, and a
`TaskInput` keeps binding as one whole input. The spec is what Airflow
records and shows.
Since the spec travels as JSON, a literal with no JSON form is now
rejected where the Dag is parsed instead of reaching a task that cannot
receive it.
* Java SDK: Keep one unserializable Dag from failing a whole bundle's parse
A Dag that could not be serialized threw out of the parse response, so
every other Dag in the same JAR vanished with it and `import_errors` was
never filled. Report the failure under the bundle-relative path the way
the TypeScript SDK does, naming the Dag, and serialize the rest.
* Java SDK: Mark the native Dag capabilities as shipping in 3.4
`native-dag-authoring`, `task-args`, `taskflow-dependencies` and
`task-group` are new in this release, not in 3.3, which is what the
matrix's "supported since" column says.
* Java SDK: Accept every cron schedule croniter does and bind wired inputs first
The cron shape check now accepts comma lists of names, month and weekday
names, the `#`, `L` and `W` qualifiers and the `@midnight` and `@annually`
aliases, which Python stores unexpanded. Each of these previously turned
the Dag into an import error. The schema-default comment records why
explicit default values are dropped, and the timezone TODO and the Java
docs note that a cron Dag with no start date is scheduled in UTC.
A wired input wins over a runtime binding in ArgValues. The Java docs and
ADR 0007 now say so, and a test pins it.
The conformance run now passes `--supports literal_inputs` and the Java
serializer wires each task's upstream handles and literals as call
arguments, so the run exercises `_arg_bindings`. The hook's file filter
covers `internal/Fields.kt`. Selective checks skip the hook unless a
java-sdk file, Airflow's serializer or schema, or the shared harness
changed, through a new file group because serializer changes do not force
full tests.
* Java SDK: Group a native Dag's tasks with task groups
* Java SDK: Reserve the whole group view surface and pin group edge order
The reserved names of a group's wiring view are now read from Deps.TaskGroup, so
endpoints, before and after are rejected as task method names inside a group, not
only groupId and nodes. A task named before would otherwise win overload
resolution over the inherited before(Flow...) and silently register a task.
Add tests for that clash, for duplicate group IDs in one scope, and for the order
group edges resolve in: an empty group prefers an earlier edge over its parent,
and edges resolve in the order they were drawn.
Correct the Task groups docs and TaskGroupRef KDoc: only what a group holds is
read once, while edges still resolve in drawing order. Restore the missing
transform, load and audit calls in the group wiring example so it parses.
* Add task state store API to the Java SDK
Airflow 3.3 added a task state store (AIP-103) and the supervisor already
handles the Get/Set/Delete/ClearTaskStateStore messages, but the Java SDK
had no task-facing API for it and capabilities.yaml listed it as
unsupported. Java tasks could not keep state such as an external job ID
across retries.
* Expose taskStateStore as a getter and fix the retry scope wording
Review found that a @JvmField on Client cannot be stubbed by Mockito, so a
Java unit test of a task that reads the store hit a NullPointerException.
The docs also said entries survive later runs, but the store is scoped to
one task instance, so they only survive retries within the same Dag run.
* Apply the default retention to task state keys stored from Java
A key stored without a retention never expired, while Python applies
[state_store] default_retention_days. The coordinator passes that setting
to the JVM as AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS, so read it
the same way the Go SDK does: absent means 30 days, 0 means never expire,
and a malformed value fails instead of retaining for a different period.
TaskStateStore.NEVER_EXPIRE replaces the old "no retention" meaning, and a
zero or negative retention is rejected.
* Fail instead of guessing when the default retention is not passed
Review on the Go and TS task state store PRs asked to drop the hardcoded
30-day fallback: the coordinator always passes
AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS, so a missing variable means
the JVM was not launched by the coordinator, and guessing a period there
would drift from the deployment's config. Also leave a TODO for the
max_value_storage_bytes warning the Python accessor emits.
* Java SDK: Declare a Dag's task graph with @Builder.Deps
An annotated Dag class could name its tasks but not say how they connect,
so a Dag whose annotations carried everything else still needed a Python
Dag file for its graph. Declaring the graph by calling the tasks lets
javac check that each task is fed something its parameter accepts.
Bundle.register(Class) now also takes the @Builder.Dag class itself and
registers the Dag its generated builder builds, so there is still only one
name to keep in sync.
* Java SDK: Quote wired task ids as Java string literals
* Java SDK: Validate the @Builder.Deps class at compile time
* Java SDK: Reject task method names that clash with the wiring view
* Java SDK: Type a wired numeric input by its parameter's own type
* Java SDK: Register both the Dag and handlers of a class, unwrapping builder failures
* Java SDK: Reject a raw null wiring argument and test the recorder's guards
* Java SDK: Reword the native-Dag docs without em-dashes
* Java SDK: Drop Bundle tests that repeat existing coverage
* Java SDK: Require a @Builder.Deps class on every @Builder.Dag class
* Java SDK: Back the k8s test's stub tasks with task handlers
* Java SDK: Reject overloaded task and task-handler methods
* Java SDK: Fix the Dag examples a required wiring class invalidates
A core Dag-serialization schema change shipped with the Go SDK's vendored copy
left behind, uncaught by any hook: the sync hooks for Go and Java were manual
only, and the one check that runs on every commit watched the vendored copies,
not the Python source, so it never saw the schema change at all. The same
drift already caused a two-release-old bug in the Go SDK's generated models
(#73954).
Go's and Java's sync hooks now run on every commit and are triggered by the
Python source too, the way the TypeScript SDK's already was, refreshing the
vendored copy and failing when they had to. Java's Gradle task keeps the same
task name and stays silent for generateDagDsl's own dependency, so an
ordinary in-repo build still refreshes a schema someone is mid-edit on
without breaking; only the prek hook's invocation passes the flag that makes
it fail.
* Check that a serialized Dag can be stored and loaded
DagSerialization.validate_serialized_dag checks the JSON schema,
deserializes a copy and looks for a cycle in the task graph. It raises
DeserializationError naming the Dag, and for a cycle a task on it. It is
meant for a serialized Dag that no Python code built, such as one a
Lang-SDK runtime returns, so the Python SDK's cycle check never ran on
it. The cycle check runs the shared detect_cycle over the serialized
downstream edges.
* Fill a serialized Dag's unset settings from the Airflow config
A Lang-SDK runtime cannot read the Airflow config, so it leaves out the
Dag settings that a Python Dag reads from it when unset. The new
DagSerialization.fill_config_defaults fills max_active_tasks,
max_active_runs, max_consecutive_failed_dag_runs, catchup and
disable_bundle_versioning in a serialized Dag from the config. A value
the Dag sets is kept.
* Reject malformed task entries in a serialized Dag
The schema put additionalProperties on the tasks array, which JSON
Schema ignores, so a task entry like {} passed validation. Each entry
must now be an operator with a string task_id, and
validate_serialized_dag rejects a repeated task id. A schema error
now names the failing field's path.
Also pin both boolean config defaults to different values in the
fill_config_defaults tests, and cover the unwrapped cause of a
deserialization error.
* Check SDK Dags with the Dag processor's validation in conformance
receive() now calls DagSerialization.fill_config_defaults and
validate_serialized_dag, so the conformance job runs the same checks
as the Dag processor and the config-backed field list lives in one
place.
* Note in the parsing ADR that runtimes may omit config-backed fields
* Serialize the SDK Dag in tests that write a dag_maker Dag
These tests passed the dag_maker proxy of the serialized Dag to
LazyDeserializedDAG.from_dag, which serialized each task as its repr.
The tasks schema now rejects that, so they serialize the SDK Dag.
* Sync the Java SDK copy of the Dag schema
* Sync the Go SDK copy of the Dag schema
* Check a serialized Dag as the SDK Dag is checked and return it
validate_serialized_dag now applies the max_active_runs and catchup rules the
SDK Dag runs when it is built, to the Dag it deserializes. The cycle check
walks that Dag's tasks, so edges from the legacy _downstream_task_ids key and
from client_defaults count too. It returns the deserialized Dag.
* Report a schema error's path the same way under every jsonschema version
jsonschema 4.26 writes $.dag.tasks[2]['__type'] where 4.23 writes
$.dag.tasks[2].__type, so the import error and three tests changed with
the installed version. Build the path from the error's absolute path.
* Java SDK: Resolve a task's arguments from the Dag's own wiring
A Dag authored in Java has no Python call site, so the supervisor sends no
argument bindings for it and a task's data parameters had nothing to
resolve against. Where the Dag wired inputs for a task, those stand in;
a stub-backed task keeps reading its bindings, including when the call
site bound none.
The authoring surface that records those inputs lands in the next commit,
so nothing declares them yet.
* Keep taskDef out of Context's constructor and reuse resolveWiredAll
* Use ADR-0007 wording for the wired-arity warning
* Java SDK: Describe native Java Dags without reference to Python
A Dag authored entirely in Java has no Python in the picture, so teaching
its edge verbs as "Java's spelling of >> and <<" asks the reader to know a
language the feature does not involve. The mixed-language surface keeps its
Python references, because there the Python Dag file is part of what the
reader has to understand.
* Java SDK: Generate Dag and task configuration from the serialization schema
A Java-authored Dag could carry no schedule, retries, tags or any other
Airflow setting, so it could be defined but not configured. Hand-listing
those settings in Java would drift from the Python semantics they mirror,
so they are recorded and validated against Airflow's own Dag serialization
schema instead, and take effect once the runtime answers parse requests.
The `Builder` annotations are generated from a copy of that schema vendored
at java-sdk/sdk/schema/dag-schema.json, so a source release builds without
the monorepo; a prek hook keeps the copy in sync.
* Java SDK: Draw a Java-authored Dag's edges with before and after
A Dag authored in Java had no way to say what runs after what. Without a
Python Dag file to hold the graph, every task was a root, so a Dag that
Java owns end to end could not be expressed at all.
The verbs are before and after, matching the TypeScript SDK, and leaving
then free for the conditional control flow that reads as a then-branch.
They are the only way to draw an edge, so one graph has one spelling
however it was built.
* Add notes on delibrate implementation choices
---------
Co-authored-by: Tzu-ping Chung <tp@astronomer.io>
* Java SDK: Add setVariable and deleteVariable to the task Client
Signed-off-by: PoAn Yang <payang@apache.org>
* Remove null description case
Signed-off-by: PoAn Yang <payang@apache.org>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
* Java SDK: Register tasks as first-class TaskDef objects
`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.
The KubernetesExecutor lang-SDK example adopts the new names directly rather
than pinning to the old ones: CI now builds those artifacts from the branch
under test (#71527), so the example no longer has to compile against whatever
the SDK on main happens to expose.
* Java SDK: Clarify task ownership validation
* Java SDK: Simplify manual task registration
Manual interface-based bundles are authored directly by users, so requiring every call to wrap the task class in TaskDef adds noise to the common case while the SDK can retain first-class task definitions internally.
* Add Java SDK capability manifest and compatibility matrix
The Language SDK conformance spec defines which TaskInstance states and
capabilities a Language SDK may declare, but nothing lets an SDK say what it
actually supports. Readers of the Java SDK docs cannot tell which parts of the
spec the runtime implements today — and the gaps are real and moving (native Dag
authoring, deferral, and the state stores are not there yet), so an unqualified
"see the conformance spec" overstates what a Java task can do.
The manifest is hand-authored YAML rather than something derived from the SDK's
own sources: it describes the SDK instead of being part of it, so it has no
business shipping in a user's runtime dependencies, and deriving it would mean a
JDK and a Gradle run in a hook that only ever renders a table. It therefore sits
above every subproject in settings.gradle.kts, where no source set can pick it
up. Validation carries the weight a compiler would have: unknown keys are
rejected along with missing ones, since it is the only thing standing between a
typo and a wrong published table.
A prek hook regenerates the contributor-facing README table and the Dokka module
doc from the manifest, so a capability landing in the runtime is a one-line edit
rather than three tables to update by hand. The shared schema, SDK registry, and
renderer let the Go and TypeScript SDKs declare theirs the same way.
* Keep Language SDK compatibility matrices accurate
Contradictory native-Dag declarations could pass validation while being rendered as not applicable. Published matrix tables also need to remain readable without forcing unrelated headers onto one line.
Serializing each client's whole send → receive → deserialize round trip
behind one lock (#69080) allows only a single request in flight at a
time, so a task issuing concurrent calls over the shared comm socket
pays a full, serialized round trip per call.
Every frame already carries a request id, so a single background
dispatcher can read the socket and route each response by id, with the
lock guarding only the write.
Signed-off-by: PoAn Yang <payang@apache.org>
* Fix Java SDK numeric XCom casting
Generate accessors that widen through Number instead: primitive params use
((Number) x).intValue(), and boxed params use Optional.ofNullable(...).orElse(null)
so an absent XCom stays null. Add a java_xcom_casting_example DAG and processor
tests covering int->long->double widening and an absent boxed value.
Signed-off-by: PoAn Yang <payang@apache.org>
* Remove unused key field from Java SDK Builder.XCom annotation
Signed-off-by: PoAn Yang <payang@apache.org>
* Add float cross-type coverage to Java SDK XCom casting tests
Signed-off-by: PoAn Yang <payang@apache.org>
* Remove unused comment
Signed-off-by: PoAn Yang <payang@apache.org>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
* Java SDK: Serialize coordinator client send/receive for thread safety
A Java task may call the SDK client from multiple threads. Every call goes through
one request/response channel to the supervisor. So it must be thread safe.
Signed-off-by: PoAn Yang <payang@apache.org>
* Move requestId outside the lock
Signed-off-by: PoAn Yang <payang@apache.org>
---------
Signed-off-by: PoAn Yang <payang@apache.org>
PoAn Yang
Aaron Chen
Pierre Jeambrun
Jason(Zhe-You) Liu
Jarek Potiuk
ChickenBenny
Andrew Chang
Ash Berlin-Taylor
dependabot[bot]andJarek Potiuk
Tzu-ping Chung