uranusjr commented on code in PR #74701:
URL: https://github.com/apache/airflow/pull/74701#discussion_r4335749577
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -161,853 +108,675 @@ Configure the build in ``build.gradle``:
}
airflowBundle {
- mainClass = "com.mycompany.airflow.sales.Main"
+ mainClass = "com.mycompany.airflow.hello.Main"
}
-The ``annotationProcessor`` dependency is required for this annotation-based
example. Omit it when using the
-interface-based API. If the project does not have the Gradle Wrapper yet,
generate it from the project root
-with ``gradle wrapper``.
-
-Implement the Java tasks
-~~~~~~~~~~~~~~~~~~~~~~~~
-
-Save the task implementations as
-``src/main/java/com/mycompany/airflow/sales/SalesPipeline.java``:
+Declare the Dag in ``HelloDag.java``:
.. code-block:: java
- package com.mycompany.airflow.sales;
+ package com.mycompany.airflow.hello;
- import org.apache.airflow.sdk.Airflow;
+ import org.apache.airflow.sdk.*;
- public class SalesPipeline {
+ @Airflow.Dag(id = "java_hello", schedule = "@daily")
+ public class HelloDag {
- @Airflow.TaskHandler(dag = "sales_pipeline", task = "extract")
+ @Airflow.Task(queue = "java")
public long extract() {
- return 3;
+ return 42L;
}
- @Airflow.TaskHandler(dag = "sales_pipeline", task = "transform")
- public long transform(long recordCount) {
- return recordCount * 2;
+ @Airflow.Task(queue = "java")
+ public void report(long rows) {
+ System.out.println("extracted " + rows + " rows");
}
- }
-
-.. note::
-
- The graph is declared once, in the Python Dag file: ``transform(extract())``
feeds the upstream's
- return value into the downstream's parameter by calling tasks like
functions. The supervisor sends
- the resulting *argument bindings* to the Java runtime, and each Java data
parameter receives
- whatever the Python call site bound at its position — an upstream task's
XCom or an inline
- literal. See :ref:`java-sdk/arg-binding`.
-Add the Java entry point
-~~~~~~~~~~~~~~~~~~~~~~~~
+ @Airflow.Deps
+ static class Wiring implements HelloDagDeps {
+ void depends() {
+ report(extract());
+ }
+ }
+ }
-Save the entry point as
``src/main/java/com/mycompany/airflow/sales/Main.java``:
+Serve it from ``Main.java``:
.. code-block:: java
- package com.mycompany.airflow.sales;
+ package com.mycompany.airflow.hello;
import org.apache.airflow.sdk.Bundle;
import org.apache.airflow.sdk.Server;
public class Main {
public static void main(String[] args) {
- Server.create(args).serve(new Bundle().register(SalesPipeline.class));
+ Server.create(args).serve(new Bundle().register(HelloDag.class));
}
}
-``register`` takes the class the annotations are on, so there is no second
name to keep in sync: the
-annotation processor generates the registrar it reads during compilation.
-
-Build and deploy
-~~~~~~~~~~~~~~~~
-
-Run the bundle task from the ``sales-pipeline-java/`` project root:
+Build the bundle from the project root:
.. code-block:: bash
./gradlew bundle
-The deployable JAR or JARs are now in ``build/bundle/``. Copy or mount that
entire directory into a path
-available on every worker that can consume the ``java`` queue. For example,
deploy it as
-``/opt/airflow/jars/sales-pipeline/``. Java source files do not need to be
deployed to Airflow.
+Copy ``build/bundle/`` into a Dag bundle, for example a ``java-hello/``
directory in the default ``dags-folder``
+bundle, which reads the ``[core] dags_folder`` directory (see
+:doc:`Dag bundles </administration-and-deployment/dag-bundles>`). Then send
the ``java`` queue to the Java
+coordinator in ``airflow.cfg``:
-For a local deployment where those paths are writable, the copy commands could
be:
+.. code-block:: ini
-.. code-block:: bash
+ [sdk]
+ coordinators = {"java": {"classpath":
"airflow.sdk.coordinators.java.JavaCoordinator"}}
+ queue_to_coordinator = {"java": "java"}
- mkdir -p /opt/airflow/jars/sales-pipeline
- cp build/bundle/*.jar /opt/airflow/jars/sales-pipeline/
+Airflow reads the bundle like any other Dag file: ``java_hello`` shows up in
the UI, runs on its schedule,
+and ``report`` receives the value ``extract`` returned.
-Deploy ``sales_pipeline.py`` separately through the deployment's normal Dag
delivery process. For example,
-that process might sync it to ``${AIRFLOW_HOME}/dags/`` or package it in a Dag
bundle; neither location is
-inside or relative to ``sales-pipeline-java/``.
+.. _java-sdk/dag-definition:
-Configure Airflow to register the JAR directory as a Dag bundle, point the
coordinator at it, and route the
-``java`` queue to the coordinator. Add the following sections to the file
selected by ``AIRFLOW_CONFIG`` (by
-default, ``${AIRFLOW_HOME}/airflow.cfg``), or set the equivalent
``AIRFLOW__*`` environment variables:
+Declaring a Dag in Java
+-----------------------
-.. code-block:: ini
+A Dag can be declared in two styles, and each section below shows both:
- [dag_processor]
- dag_bundle_config_list = [
- {"name": "dags-folder", "classpath":
"airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
- {
- "name": "java-task-handlers",
- "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
- "kwargs": {"path": "/opt/airflow/jars/sales-pipeline"}
- }
- ]
+* **Annotations.** ``@Airflow.Dag`` marks a class as a Dag, ``@Airflow.Task``
marks its task methods, and a
+ nested ``@Airflow.Deps`` class wires them together. The annotation processor
generates the code that
+ connects them, so ``javac`` type-checks the graph.
+* **Interface.** Each task is a class implementing ``Task``, and a ``DagDef``
object registers the tasks and
+ their dependencies. It needs no annotation processor.
- [sdk]
- coordinators = {
- "java": {
- "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
- "kwargs": {"task_handler_bundle_name": "java-task-handlers"}
- }
- }
- queue_to_coordinator = {"java": "java"}
+Tasks and their inputs
+~~~~~~~~~~~~~~~~~~~~~~
-``java`` is a user-chosen coordinator name, not a reserved value. The value
assigned to the queue in
-``queue_to_coordinator`` must match a key in ``coordinators``, and
``task_handler_bundle_name`` must match a
-Dag bundle name in ``dag_bundle_config_list``. See
:ref:`java-sdk/coordinator-config` for how JARs are
-located.
+With annotations, ``@Airflow.Task`` marks a method as a task, and the nested
``@Airflow.Deps`` class declares
+the graph by calling the task methods:
-Restart the affected Airflow components after changing this configuration. The
coordinator config, the JARs
-and a JRE must be available wherever tasks execute and on the Dag processor.
With ``CeleryExecutor``, tasks
-execute on the Celery workers; with ``LocalExecutor``, they run in
subprocesses on the scheduler's host. The
-Dag processor checks the stub tasks of ``sales_pipeline.py`` against the task
handlers the JARs register, so
-it runs them too. The API server does not need any of it. Register the Dag
bundle in
-``[dag_processor] dag_bundle_config_list`` on every component, like your other
Dag bundles: the worker and
-the Dag processor resolve ``task_handler_bundle_name`` through it, and
wherever the ``[sdk]`` config is read
-it is rejected if the name is missing there. The Dag processor still receives
``sales_pipeline.py`` through
-the separate Dag delivery process.
+.. code-block:: java
-A Dag processor with this ``[sdk]`` configuration also parses the executable
JARs of every Dag bundle,
-and needs a JRE to do so (see :ref:`java-sdk/native-dag-parsing`). Keep it
from parsing a bundle's JARs
-with that bundle's ``.airflowignore``, whose pattern syntax follows ``[core]
dag_ignore_file_syntax``.
+ @Airflow.Dag(id = "java_etl", schedule = "@daily")
+ public class EtlPipeline {
-With ``task_handler_bundle_name`` set to its own bundle, as above,
``java-task-handlers`` holds only the
-JARs the Python Dag's tasks run, so ignoring everything in it is safe:
+ @Airflow.Task(retries = 2)
+ public long extract() {
+ return 42L;
+ }
-.. code-block:: bash
+ @Airflow.Task
+ public long transform(long extracted, double factor) {
+ return (long) (extracted * factor);
+ }
- # [core] dag_ignore_file_syntax = glob (the default)
- echo '*' > /opt/airflow/jars/sales-pipeline/.airflowignore
+ @Airflow.Task
+ public void load(long transformed) {}
- # [core] dag_ignore_file_syntax = regexp; a bare * is not a valid pattern
and is dropped
- echo '.' > /opt/airflow/jars/sales-pipeline/.airflowignore
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ load(transform(extract(), lit(0.9)));
+ }
+ }
+ }
-With ``task_handler_bundle_name`` unset, the JAR sits in the same Dag bundle
as ``sales_pipeline.py``, so
-ignore only the JAR, not the whole bundle:
+The annotation processor generates the ``EtlPipelineDeps`` interface, with one
method per task. Passing the
+value one call returned into another call makes the second task wait for the
first and receive its value.
+``lit(...)`` passes a constant instead, which must be a string, number,
boolean, list or map.
-.. code-block:: bash
+* Every task method is called in the wiring class. A task the wiring leaves
out fails when the Dag is read.
+* A task method that takes arguments is called once; keep its result in a
local variable to use it twice.
+* Values move between tasks as XComs, so a task returns data JSON can hold
(see :ref:`java-sdk/types`).
- # [core] dag_ignore_file_syntax = glob (the default)
- echo '*.jar' >> /opt/airflow/dags/.airflowignore
+With the interface, each task is a class, and ``dag.task(taskId, TaskClass)``
registers it. A task reads the
+value of an upstream task itself, through ``client.getXCom``:
- # [core] dag_ignore_file_syntax = regexp
- echo '\.jar$' >> /opt/airflow/dags/.airflowignore
+.. code-block:: java
-After Airflow has parsed the Dag, trigger it from the UI or command line:
+ public class Extract implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ client.setXCom(42L);
+ }
+ }
-.. code-block:: bash
+ public class Load implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ var extracted = ((Number) client.getXCom("extract")).longValue();
+ }
+ }
- airflow dags trigger sales_pipeline
+ var dag = new DagDef("java_etl").config("schedule", "@daily");
+ var extract = dag.task("extract", Extract.class).config("retries", 2);
+ var load = dag.task("load", Load.class);
+ extract.before(load);
-The Java ``extract`` task returns ``3``, ``transform`` returns ``6`` through
XCom, and the Python ``load`` task
-logs ``Loaded: 6``.
+A task class must be ``public`` and concrete, have a public no-argument
constructor, and be ``static`` if it is
+nested in another class. The runtime creates a new instance for every task run.
-See :ref:`java-sdk/coordinator-config` for the full list of accepted
``kwargs``.
-For a larger example that exercises connections, variables, logging, and both
task APIs, see the
-`Java SDK example in the Airflow repository
<https://github.com/apache/airflow/tree/main/java-sdk/example>`__.
+Register either kind of Dag on the ``Bundle`` that ``main`` serves:
-Writing tasks
--------------
+.. code-block:: java
-The Java SDK offers two APIs for implementing tasks. Both produce the same
runtime behavior; the choice is a
-matter of style.
+ Server.create(args).serve(new
Bundle().register(EtlPipeline.class).register(dag));
-.. _java-sdk/annotation-api:
+Task ids
+~~~~~~~~
-Annotation-based API
-~~~~~~~~~~~~~~~~~~~~
+``@Airflow.Task`` takes the task id from the method name unless ``id`` is set,
and ``@Airflow.Dag`` takes the
+Dag id from the class name unless ``id`` is set. ``dag.If`` and ``dag.Switch``
take it from the class name unless
+you pass one.
-Annotate a plain Java class and let the SDK generate the boilerplate at
compile time.
+Dag and task options
+~~~~~~~~~~~~~~~~~~~~
-.. list-table::
- :header-rows: 1
- :widths: 30 70
-
- * - Annotation
- - Purpose
- * - ``@Airflow.TaskHandler(dag = "...", task = "...")``
- - Marks a method as the Java body of a task the Python Dag file declares
with ``@task.stub``.
- ``dag`` must match the ``dag_id`` and ``task`` the stub function name;
omitting ``task``
- derives it from the method name. There is no class-level annotation on
this surface — the
- Dag is the Python file's, so the handler names the pair it binds to.
- * - ``@Airflow.Dag(id = "...")``
- - Marks a class as a Dag that Java itself owns. Attributes
(``schedule``, ``description``,
- ``tags``, ``catchup``, …) are Airflow's own Dag settings; only
attributes written
- explicitly are applied. See :ref:`java-sdk/native-dags`.
- * - ``@Airflow.Task(id = "...")``
- - Marks a method as a task of a Java-owned Dag. If ``id`` is omitted the
method name is
- used. Further attributes (``retries``, ``queue``, ``retryDelay``, …)
are Airflow's own
- task settings; only attributes written explicitly are applied.
- * - ``@Airflow.Deps``
- - Marks the nested class that declares the task graph in Java,
TaskFlow-style. Required for
- a Dag that Java owns end to end. See :ref:`java-sdk/native-dags`.
- * - ``TaskInput`` / ``@ArgName("...")``
- - Marks a class as a task's input, so keyword arguments bind by name
instead of by position:
- each public field receives the argument whose name matches it, ignoring
case and
- underscores. ``@ArgName`` pins a name the match cannot reach, or
renames the argument
- deliberately. See :ref:`java-sdk/arg-binding`.
-
-Besides the annotations, a task method may declare a ``Client`` and a
``Context`` parameter in any
-position; the SDK injects both. Every other parameter is a *data parameter*
and receives an
-argument bound by the Python ``@task.stub`` call site.
-
-The annotation processor generates a ``<ClassName>Builder`` class that wires
up the task
-registry and resolves data parameters and XCom pushes automatically.
+The attributes of ``@Airflow.Dag`` and ``@Airflow.Task``, and the keys of
``DagDef.config`` and
+``TaskDef.config``, are Airflow's own Dag and task settings. Annotation
attributes are ``camelCase``
+(``retryDelay``), and ``config`` keys are spelled as in Python
(``"retry_delay"``). Only the settings you write
+are applied, so Airflow's defaults hold for the rest.
.. code-block:: java
+ @Airflow.Dag(id = "java_etl", schedule = "@daily", startDate =
"2026-01-01T00:00:00Z", catchup = false)
- public class MyDag {
+ @Airflow.Task(retries = 2, retryDelay = "PT5M", queue = "java")
- @Airflow.TaskHandler(dag = "my_dag", task = "fetch")
- public String fetch(Client client) throws Exception {
- var conn = client.getConnection("my_api");
- // implement task logic
- return result;
- }
-
- @Airflow.TaskHandler(dag = "my_dag", task = "process")
- public long process(Client client, String fetched) {
- var threshold = (String) client.getVariable("process_threshold");
- // implement task logic
- return count;
- }
- }
-
-A task method may declare ``throws Exception``; any uncaught exception causes
the task instance to be marked
-as failed in Airflow (triggering retries if configured on the stub).
+.. code-block:: java
-.. _java-sdk/interface-api:
+ var dag = new DagDef("java_etl").config("schedule",
"@daily").config("catchup", false);
+ dag.task("load", Load.class).config("retry_delay",
Duration.ofMinutes(5)).config("queue", "java");
-Interface-based API
-~~~~~~~~~~~~~~~~~~~
+Durations and date-times are ISO-8601 strings in annotations, and
``java.time.Duration`` and
+``java.time.OffsetDateTime`` values in ``config``. An unknown setting or a
value of the wrong type fails the
+build for an annotation, and the ``config`` call for an object.
-Implement the ``Task`` interface directly for full control over how tasks are
registered and how XComs are
-read. Each task is registered as a ``TaskDef`` on a ``DagDef``; both carry a
fluent
-``config(key, value)`` whose keys are Airflow's own setting names.
+A cron ``schedule`` runs in the time zone of ``startDate``, or in UTC without
one.
-The runner creates a fresh instance of the task class through reflection for
every task-instance run,
-which puts four constraints on the class:
+Every task needs a queue that :ref:`queue_to_coordinator
<java-sdk/coordinator-config>` sends to a
+``JavaCoordinator``. There is no Dag-level queue yet, so set ``queue`` on each
task.
-* The task class itself must be ``public``.
-* It must be concrete: not abstract and not an interface.
-* It must declare a public no-argument constructor.
-* If nested inside another class, it must be a ``static`` nested class.
+Order-only dependencies
+~~~~~~~~~~~~~~~~~~~~~~~
-A class that violates any of these fails at runtime with a ``Cannot
instantiate task class`` error in the
-task log.
+When a task only has to run after another, without receiving its value, draw
the dependency with ``before``
+and ``after``, the Java spelling of Python's ``>>`` and ``<<``. Both work the
same in the two styles:
.. code-block:: java
- import org.apache.airflow.sdk.*;
+ rows.before(audit()); // rows >> audit
+ load.after(transform, validate); // [transform, validate] >> load
- public class FetchTask implements Task {
- @Override
- public void execute(Context context, Client client) throws Exception {
- var conn = client.getConnection("my_api");
- // implement task logic
- client.setXCom(result);
- }
- }
+Both take any number of tasks and return the task they were called on, so
calls chain. ``Flow.of(a, b).before(c,
+d)`` draws every edge between two sets in one call. ``Flow`` is
``org.apache.airflow.sdk.Deps.Flow``, and needs
+no import inside the wiring class. An edge to a task of another Dag, or one
that closes a
+cycle, fails when the Dag is registered.
-Implement ``InputTask<I>`` instead when the Python Dag calls the stub with
TaskFlow arguments: the SDK
-resolves them from the call site and passes them in. The type argument is a
``TaskInput`` whose public
-fields declare what the task expects. See :ref:`java-sdk/arg-binding`.
+Task groups
+~~~~~~~~~~~
-Register each task against the Dag and task the Python file declares.
``register`` is one
-overloaded verb: these ids, or the class an annotated handler lives on.
+A task group gathers tasks that the Airflow UI shows as one node, as Python's
``TaskGroup`` does. Every id
+declared in a group is prefixed with the group's id.
+
+With annotations, a ``static`` nested ``@Airflow.TaskGroup`` class holds the
tasks of one group, and nesting
+one group class in another nests the groups:
.. code-block:: java
- public class MyBundle {
- public static class ProcessInput implements TaskInput {
- public String fetched;
- }
+ @Airflow.TaskGroup // group "Staging", from the
class name
+ static class Staging {
+ @Airflow.Task
+ public long stage(long rows) { ... } // task "Staging.stage"
- public static class ProcessTask implements InputTask<ProcessInput> {
- @Override
- public void execute(Context context, Client client, ProcessInput
input) throws Exception {
- // implement task logic
- client.setXCom(input.fetched);
- }
+ @Airflow.TaskGroup(id = "checks")
+ static class Checks {
+ @Airflow.Task
+ public void nulls(long staged) { ... } // task "Staging.checks.nulls"
}
+ }
- public static void main(String[] args) {
- var bundle = new Bundle()
- .register("my_dag", "fetch", FetchTask.class)
- .register("my_dag", "process", ProcessTask.class);
- Server.create(args).serve(bundle);
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ var staged = staging().stage(extract());
+ staging().checks().nulls(staged);
}
}
-A task class can be top-level like ``FetchTask``, or a nested ``static`` class
like ``ProcessTask``.
-Place them under the standard ``src/main/java/<package>/`` source tree, and set
-``airflowBundle.mainClass`` to the class that provides ``main``. From that
point onward, both APIs
-use the same ``./gradlew bundle`` command and deploy the resulting
``build/bundle/`` directory in the same way.
+With the interface, ``taskGroup`` declares a group on the Dag or on another
group:
-See the `Java SDK API Reference
<https://airflow.apache.org/docs/java-sdk/stable/>`__ for more details.
+.. code-block:: java
-.. _java-sdk/arg-binding:
+ var staging = dag.taskGroup("staging");
+ var stage = staging.task("stage", Stage.class); //
"staging.stage"
+ staging.taskGroup("checks").task("nulls", Nulls.class).after(stage); //
"staging.checks.nulls"
+ extract.before(staging);
-Binding stub arguments
-~~~~~~~~~~~~~~~~~~~~~~
+A group takes ``before`` and ``after`` too, which orders the whole group. A
group id is made of ASCII letters,
+digits, underscores and dashes, and no task or other group in the Dag can
share it.
-Calling a ``@task.stub`` TaskFlow-style in the Python Dag is what declares the
graph, and the
-supervisor delivers the resulting argument bindings to the Java runtime with
every task run. A
-binding carries either an upstream task's ``return_value`` XCom or an inline
literal written at the
-call site.
+Conditional branching
+~~~~~~~~~~~~~~~~~~~~~
-Positional binding
-^^^^^^^^^^^^^^^^^^
+A condition is a task that returns a boolean, and names the task each outcome
runs. The side not taken is
+skipped.
-A task method's data parameters bind **by position**, in declaration order —
the injected ``Client``
-and ``Context`` parameters do not take up a position. Java parameter names
are not part of the API,
-so renaming one in an IDE never rebinds an input.
+With annotations, mark the method with ``@Airflow.If``, and name the sides in
the wiring class:
-.. code-block:: python
+.. code-block:: java
- @task.stub(queue="java")
- def score(rows, threshold): ...
+ @Airflow.If
+ public boolean hasRows(long transformed) {
+ return transformed > 0;
+ }
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ var transformed = transform(extract(), lit(0.9));
+ hasRows(transformed).Then(load(transformed)).Else(reportEmpty());
+ }
+ }
- score(load_rows(), 0.75)
+With the interface, implement ``ConditionTask`` and register it with
``dag.If``:
.. code-block:: java
- @Airflow.TaskHandler(dag = "etl", task = "score")
- public long score(Client client, long rows, double threshold) {
- // rows <- the load_rows XCom (position 0)
- // threshold <- the literal 0.75 (position 1)
+ public class HasRows implements ConditionTask {
+ @Override
+ public boolean decide(Context context, Client client) {
+ return ((Number) client.getXCom("transform")).longValue() > 0;
+ }
}
-A primitive parameter cannot hold ``null``, so the task fails with
``MissingXComException`` when its
-binding resolves to nothing; declare a boxed type (``Long``, ``Double``, …) to
receive ``null``
-instead. The method must declare as many data parameters as the call site
bound: positions carry
-the whole meaning of a flat binding, so any other count has already shifted
them, and the task fails
-rather than running on arguments it has mistaken for others. A parameter the
Python call omitted
-does not count towards that. Its default still arrives, but a method that
does not declare it is
-not reading shifted arguments, so the SDK drops it before comparing the two
counts.
-
-Generic parameters are decoded element by element. Declare ``List<Double>``
and ``values.get(0)``
-really is a ``Double``, even though the call site passed whole numbers and the
wire carries them as
-integers. Without that the list would hold ``Long`` values while claiming to
hold ``Double``, and
-the ``ClassCastException`` would land on whichever line first read an element
rather than on the
-binding that got it wrong.
-
-Named binding with a ``TaskInput``
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-
-To bind keyword arguments by name, declare a class implementing ``TaskInput``.
Each of its public
-non-final fields receives the argument whose name matches it, **ignoring case
and underscores**, so
-the stub's ``snake_case`` arguments reach ``camelCase`` Java fields with
nothing declared. That is
-the same fold the Go and TypeScript SDKs apply, so one Python signature binds
identically in every
-SDK.
-
-The class needs a public no-argument constructor. A name mismatch in either
direction is logged
-rather than failed, because a field binds by name: a field nothing supplies
keeps its Java default,
-and an argument no field claims changes nothing the task reads. Once a field
has claimed its
-argument, an argument that resolves to nothing is a value and not a mistake,
so a boxed or reference
-field takes ``null`` and a primitive field fails.
-
-No two fields may claim argument names that differ only in case or
underscores; that fails when the
-bundle is built, because the fold cannot tell them apart. Two *arguments*
that collide that way
-reach only a field naming one of them exactly, and a field that would match
both is left unfilled
-and reported rather than handed the wrong value.
+ dag.If(HasRows.class).after(transform).Then(load).Else(reportEmpty);
-.. code-block:: python
+``Else`` is optional.
- @task.stub(queue="java")
- def score(region_code, threshold): ...
+Only the task on the side not taken is skipped, and it stays skipped if it is
cleared later. A task that runs
+after both sides needs a trigger rule that accepts a skipped upstream, such as
``none_failed_min_one_success``.
+Multi-way branching
+~~~~~~~~~~~~~~~~~~~
- score(region_code="emea", threshold=load_threshold())
+A switch is a task that chooses one of several tasks to run, and skips every
other one. It returns the class
+of the task to run.
-.. code-block:: java
+With annotations, mark the method with ``@Airflow.Switch``, return a class the
annotation processor generated
+for a task of the Dag, and list the cases in the wiring class:
- public static class ScoreInput implements TaskInput {
- // Pinned so the field can be called region. Or drop the annotation and
- // write: public String regionCode;
- @ArgName("region_code")
- public String region;
+.. code-block:: java
- public double threshold; // binds threshold
+ @Airflow.Switch
+ public Class<? extends Task> pickReport(long transformed) {
+ return transformed > 100 ? EtlPipelineBuilder.ReportLong.class :
EtlPipelineBuilder.ReportShort.class;
}
- @Airflow.TaskHandler(dag = "etl", task = "score")
- public long score(Client client, ScoreInput input) { ... }
-
-Reach for ``@ArgName`` when the argument name is not a legal or usable Java
identifier — a Python
-keyword such as ``class``, say — or when the field should read differently
from the argument. A
-pinned name is matched as written, with no folding.
-
-A task method declares flat data parameters **or** one ``TaskInput``, never
both, so field names and
-flat positions cannot shift each other. Mixing them, or declaring two, fails
the build.
-
-Binding in the interface-based API
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+ @Airflow.Deps
+ static class Wiring implements EtlPipelineDeps {
+ void depends() {
+ pickReport(transform(extract(),
lit(0.9))).Case(reportLong()).Case(reportShort());
+ }
+ }
-A task written against the interface has no parameter list to bind, so it
declares its input as the
-type argument of ``InputTask<I>`` instead — the same ``TaskInput`` an
annotated task would declare,
-binding the same way:
+With the interface, implement ``SwitchTask`` and register it with
``dag.Switch``:
.. code-block:: java
- public class ScoreTask implements InputTask<ScoreInput> {
+ public class PickReport implements SwitchTask {
@Override
- public void execute(Context context, Client client, ScoreInput input)
throws Exception {
- // input.region, input.threshold
+ public Class<? extends Task> choose(Context context, Client client) {
+ return ((Number) client.getXCom("transform")).longValue() > 100 ?
ReportLong.class : ReportShort.class;
}
}
-A ``TaskInput`` is the only way an interface task receives bound values; there
is no positional form.
-A positional read is safe in the code the annotation processor writes, because
it type-checks each
-position against the method signature it serves, and is the wrong thing to ask
of code a person
-writes and later reads. A call site with a single argument is worth the
one-field class.
+
dag.Switch(PickReport.class).after(transform).Case(reportLong).Case(reportShort);
-An ``InputTask`` whose type argument is not a concrete ``TaskInput`` fails
when the bundle is built,
-rather than mid-run. Plain ``Task`` remains the right interface for a task
the Dag file
-calls with no arguments.
+Exactly one case runs. A switch that returns a class that is not one of its
cases fails, and two cases cannot
+run the same class. A task that runs after several cases needs a trigger rule
that accepts a skipped upstream,
+such as ``none_failed_min_one_success``.
-.. _java-sdk/native-dags:
+Edge labels
+~~~~~~~~~~~
-Native Java Dags
-----------------
+``Label`` puts a label on an edge, which the Airflow UI shows on the graph, as
Python's ``Label`` does.
+``Label.of(text).on(...)`` goes in ``before`` or ``after`` in place of the
task or group at the other end of the edge,
+and works the same in the two styles:
-A Dag can also be authored entirely in Java: the annotations (or the
``DagDef`` / ``TaskDef``
-objects) carry the configuration, and Java declares the graph.
+.. code-block:: java
-Building the Dag in Java
-~~~~~~~~~~~~~~~~~~~~~~~~
+ loaded.before(Label.of("when empty").on(reportEmpty)); //
load >> Label("when empty") >> report_empty
+ cleaned.after(Label.of("always").on(loaded)); //
load >> Label("always") >> cleanup
+ loaded.before(Label.of("forks").on(reportEmpty, reportSuspicious)); //
one label on two edges
-``dag.task(...)`` registers a task as it creates it and hands back a handle.
``before`` and
-``after`` draw every edge on this surface, and the task body moves the data
itself, by reading the
-upstream's XCom through ``Client``:
+A label goes only on the edges between the receiver of ``before`` or ``after``
and the tasks passed to ``on``, so each
+edge of a fan-out can carry its own label. Declaring an edge that already
exists only adds the label, which is how the
+sides of a condition get theirs. A new label replaces the old one:
.. code-block:: java
- var dag = new DagDef("java_etl");
+ var hasRows =
dag.If(HasRows.class).after(transform).Then(load).Else(reportEmpty);
+ hasRows.before(Label.of("rows found").on(load), Label.of("no
rows").on(reportEmpty));
- var extract = dag.task("extract", Extract.class);
- var transform = dag.task("transform", Transform.class);
- var load = dag.task("load", Load.class);
+``Label`` is ``org.apache.airflow.sdk.Label``. Unlike ``Flow``, the wiring
class does not inherit it, so it needs an
+import. Pass the value ``on`` returns only to ``before`` and ``after``:
calling ``before`` or ``after`` on it fails. On
+an edge to or from a task group, the UI shows the label on the group's node.
- transform.after(extract).before(load);
+A label never changes which tasks an edge connects. Python's ``Label`` can:
when the end on its left is in a task group
+and the other end is not in that group, Python replaces that end with the
group.
-Both are variadic, so ``a.before(b, c)`` fans out and ``d.after(b, c)`` fans
in, and both return
-their own receiver, so a chain reads from one task outwards. ``Flow.of(a,
b).before(c, d)``, from
-``org.apache.airflow.sdk.Deps.Flow``, draws every edge between two sets in one
call.
-
-Edges are checked when the Dag is registered with a ``Bundle``: an upstream
that belongs to another
-Dag, or to no Dag, and a cycle anywhere in the graph both fail there rather
than at the first task
-run.
-
-Wiring the graph with ``@Airflow.Deps``
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+Triggering another Dag
+~~~~~~~~~~~~~~~~~~~~~~
-For a Dag written with annotations, the graph is declared by a nested
``@Airflow.Deps`` class. The
-annotation processor generates a ``<ClassName>Deps`` interface, the *wiring
view*, with one method
-per ``@Airflow.Task`` method: the injected ``Client`` and ``Context``
parameters are dropped, each
-data parameter becomes an ``Arg<T>``, and the return value becomes a
``TaskRef<T>``. Calling a view
-method registers its task, and passing the handle one returned into another
feeds the upstream's
-output into the downstream's parameter *and* wires the data edge. The call
graph is the task graph,
-and ``javac`` type-checks it.
+``TriggerDagRun`` makes a task that starts a run of another Dag, as Python's
``TriggerDagRunOperator`` does.
-Declare the wiring class as a ``static`` nested class of the Dag class that
``implements`` the
-generated view, with a no-argument ``depends()`` method:
+With annotations, a ``@Airflow.Task`` method returns one. The method runs when
the Dag is built, not when the
+task runs, so it takes no arguments:
.. code-block:: java
- @Airflow.Dag(
- id = "java_etl",
- schedule = "@daily",
- description = "Pure-Java Dag built with annotations",
- tags = {"example", "java-sdk"})
- public class EtlPipeline {
-
- @Airflow.Task(id = "extract", retries = 2)
- public long extract() {
- return 42L;
- }
-
- @Airflow.Task(id = "transform")
- public long transform(long extracted, double factor) {
- return (long) (extracted * factor);
- }
+ @Airflow.Task
+ public TriggerDagRun triggerDownstream() {
+ return new TriggerDagRun("downstream_etl").config("conf",
Map.of("source", "java_etl"));
+ }
- @Airflow.Task(id = "load")
- public void load(long transformed) {
- // implement task logic
- }
+With the interface, pass one to ``dag.task`` in place of a task class:
- @Airflow.Task(id = "audit")
- public void audit() {
- // side effect only, no data in or out
- }
+.. code-block:: java
- @Airflow.Deps
- static class Wiring implements EtlPipelineDeps {
- void depends() {
- var rows = extract();
- load(transform(rows, lit(0.9)));
- rows.before(audit()); // ordering-only edge: audit waits for extract
- }
- }
- }
+ dag.task("trigger_downstream", new
TriggerDagRun("downstream_etl").config("wait_for_completion", true));
-Every ``@Airflow.Task`` method must be called in the wiring class; a task the
wiring missed fails at
-Dag-parse time. ``lit(...)`` wires an inline constant where no upstream feeds
a parameter. A bare
-``double`` cannot be an ``Arg``, so a constant is wrapped. Airflow records
what each task is called
-with, and that record travels as JSON, so a constant has to be a string,
number, boolean, list or
-map. A view method that takes no arguments
-returns the same handle every time, so it names one node wherever it appears;
one that takes
-arguments is called once, and the wiring fails if it is called again with
arguments, so hold its
-handle in a local and reuse that.
+``config`` takes the parameters of ``TriggerDagRunOperator``, spelled as in
Python, such as ``conf``,
+``wait_for_completion``, ``deferrable``, ``poke_interval``, ``allowed_states``
and ``failed_states``. Values are
+sent as written, so Jinja templates are not rendered.
-``before``, ``after``, ``Flow.of`` and ``Label`` work here exactly as they do
on the interface
-surface. Inside the wiring class, ``Flow`` is inherited by simple name, so it
needs no import and
-never collides with ``java.util.concurrent.Flow``. ``Label`` is not
inherited, so the wiring class
-has to import it with ``import org.apache.airflow.sdk.Label;``.
+The task runs no Java code of yours, takes no inputs and cannot be a case of a
switch. With ``deferrable``, it
+waits in the triggerer, which needs the standard provider installed.
-Every ``@Airflow.Dag`` class declares a wiring class, because the graph is
what the Dag owns. A
-class that supplies only task bodies, for a Dag a Python file declares, carries
-``@Airflow.TaskHandler`` instead and contributes no Dag.
+A complete example
+~~~~~~~~~~~~~~~~~~
-.. note::
+``java-sdk/example/src/java/org/apache/airflow/example/nativedag/`` declares
the same Dag in both styles, with a
+task group, a condition with labeled edges, a switch and a triggered Dag run.
- A native Java Dag binds its task arguments from its own wiring, and the
``_arg_bindings`` it
- serializes are what Airflow records and shows. Runtime bindings (see
:ref:`java-sdk/arg-binding`)
- are what an ``@Airflow.TaskHandler`` class reads, for a task whose Dag a
Python file declares.
+.. _java-sdk/stub-tasks:
-Task groups
-~~~~~~~~~~~
+Implementing stub tasks of a Python Dag
+---------------------------------------
-A task group gathers tasks that the Airflow UI shows as one node, as Python's
``TaskGroup`` does.
-Everything declared in a group carries the group's ID as a prefix, so task
``stage`` in group
-``staging`` is the task ``staging.stage``. On the interface surface,
``taskGroup`` declares a group on
-the Dag or inside another group, and the group declares its tasks:
+Here a Python Dag declares the tasks and their dependencies, and Java
implements some of them. Use it when the
+Dag needs something a Java Dag cannot express yet (see
:ref:`java-sdk/limitations`), or to mix Python and Java
+tasks in one Dag.
-.. code-block:: java
+The Python Dag
+~~~~~~~~~~~~~~
- var staging = dag.taskGroup("staging");
- var stage = staging.task("stage", Stage.class); //
"staging.stage"
- staging.taskGroup("checks").task("nulls", Nulls.class).after(stage); //
"staging.checks.nulls"
- extract.before(staging);
+.. code-block:: python
-With annotations, an ``@Airflow.TaskGroup`` class holds the tasks of one
group, and nesting one in
-another nests the groups:
+ from airflow.sdk import dag, task
-.. code-block:: java
- @Airflow.TaskGroup // the group "Staging", after the
class
- static class Staging {
- @Airflow.Task
- public long stage(long rows) { ... } // the task "Staging.stage"
+ @dag
+ def sales_pipeline():
+ @task.stub(queue="java")
+ def extract(): ...
- @Airflow.TaskGroup(id = "checks")
- static class Checks {
- @Airflow.Task
- public void nulls(long staged) { ... } // "Staging.checks.nulls"
- }
- }
+ @task.stub(queue="java")
+ def transform(extracted): ...
- @Airflow.Deps
- static class Wiring implements EtlPipelineDeps {
- void depends() {
- var rows = extract();
- load(transform(rows, lit(0.9)));
- rows.before(audit());
- var staged = staging().stage(rows);
- staging().checks().nulls(staged);
- extract().before(staging());
- }
- }
+ @task
+ def load(transformed):
+ print(f"Loaded: {transformed}")
-The generated view nests the same way, so a group is both the namespace of
what it holds and a point
-in the flow: ``staging().checks().nulls(staged)`` reaches a task, and
``extract().before(staging())``
-orders the whole group. Task method names scope to their own group, so two
groups can each declare
-``run()``. A group class is ``static``, non-private, and needs a no-argument
constructor, because the
-generated task bodies instantiate it.
+ load(transform(extract()))
-A group stands at either end of ``before``, ``after`` and ``Flow.of``. As an
upstream it stands for
-its leaves, the tasks nothing else in the group runs after; as a downstream,
for its roots, the tasks
-that run after nothing else in the group. A group ID contains only ASCII
letters, digits,
-underscores, or dashes, and no task or other group in the Dag can share it.
-.. note::
+ sales_pipeline()
- What a group holds is read once, when the Dag is serialized, which is what
lets the wiring class
- above order a whole group before any of its tasks are declared, as
``extract().before(staging())``
- does. Python instead reads it at each ``>>``. Edges are still resolved
in the order they were
- drawn, as Python resolves them, so drawing an inner edge before or after
an outer one gives
- different upstreams.
+``@task.stub`` declares a task that is implemented elsewhere, and ``queue``
sends it to the Java coordinator.
-Edge labels
-~~~~~~~~~~~
+The Java side
+~~~~~~~~~~~~~
-An edge can carry a label. The Airflow UI shows the label on the edge in the
graph, as it does for
-Python's ``Label``. ``Label`` is a class in the ``org.apache.airflow.sdk``
package. ``Label.of`` takes
-the label text, and ``on`` takes the task or group at one end of the edge you
want to label. Pass the
-result to ``before`` or ``after`` like any other point in the flow. The text
then sits between the two
-tasks in the code, as the label sits between them in the graph:
+With annotations, ``@Airflow.TaskHandler`` marks a method as the body of a
stub task, naming the ``dag_id`` and
+``task_id`` it implements. ``task`` defaults to the method name:
.. code-block:: java
- loaded.before(Label.of("when empty").on(reportEmpty)); // load >>
Label("when empty") >> report_empty
- cleaned.after(Label.of("always").on(loaded)); // load >>
Label("always") >> cleanup
+ public class SalesPipeline {
-A label goes on edges, not on tasks. It covers only the edges between the
receiver of ``before`` or
-``after`` and the tasks and groups passed to the label's ``on``. So each edge
of a fan-out can carry
-its own label. ``on`` also takes several tasks or groups. In that case, the
label goes on the edge
-to or from each of them:
+ @Airflow.TaskHandler(dag = "sales_pipeline", task = "extract")
+ public long extract() {
+ return 3;
+ }
-.. code-block:: java
+ @Airflow.TaskHandler(dag = "sales_pipeline", task = "transform")
+ public long transform(long extracted) {
+ return extracted * 2;
+ }
+ }
- checked.before(Label.of("rows found").on(processed), Label.of("no
rows").on(reportEmpty));
- // load >> Label("forks") >> [report_empty, report_suspicious]
- loaded.before(Label.of("forks").on(reportEmpty, reportSuspicious));
+ Server.create(args).serve(new Bundle().register(SalesPipeline.class));
-When an edge already exists, declaring it again with a label only adds the
label. That is how the
-sides of a condition get their labels:
+With the interface, implement ``Task`` and register the class against the Dag
and task ids:
.. code-block:: java
- var hasRows = dag.If(HasRows.class).Then(load).Else(loadEmpty);
- hasRows.before(Label.of("rows found").on(load), Label.of("no
rows").on(loadEmpty));
-
-When an edge already has a label, a new label replaces it. Use the value that
``on`` returns only in
-the arguments of ``before`` and ``after``. Calling ``before`` or ``after`` on
that value fails. When
-an edge goes to or from a task group, the UI draws it to or from the group's
own node and shows the
-label there.
+ public class Extract implements Task {
+ @Override
+ public void execute(Context context, Client client) {
+ client.setXCom(3L);
+ }
+ }
-.. note::
+ Server.create(args).serve(new Bundle().register("sales_pipeline",
"extract", Extract.class));
- A label never changes which tasks an edge connects. Python's ``Label``
can change those tasks.
- It does so when the end written to the left of the ``Label`` is in a task
group and the other end
- is not in that same group. Python then replaces the left end with the
group that contains it.
- For example, if ``v`` is in group ``g`` and ``n`` is in no group, ``v >>
Label("x") >> n`` draws
- edges from the leaves of ``g`` instead of from ``v``. In Java, the end
that matches this
- left end is the receiver of ``before`` or ``after``.
+The task id must match the stub's, including any task group prefix.
``register`` can be called any number of
+times, so one bundle can implement the tasks of several Python Dags and
declare Dags of its own.
-Configuration attributes
-~~~~~~~~~~~~~~~~~~~~~~~~
+Passing arguments
+~~~~~~~~~~~~~~~~~
-The ``@Airflow.Dag`` and ``@Airflow.Task`` configuration attributes, and the
keys accepted by
-``DagDef.config`` and ``TaskDef.config``, are Airflow's own Dag and task
settings, under the names
-Airflow uses. Annotation attributes are ``camelCase`` (``retryDelay``);
``config`` keys are those
-names as Airflow writes them (``"retry_delay"``).
-Only attributes written explicitly at the use site are applied, so Airflow's
own defaults still
-apply to everything left out.
+The arguments the Python Dag passes to a stub task reach the Java handler,
either by position or by name.
-Durations and date-times are ISO-8601 strings in annotations (``retryDelay =
"PT5M"``,
-``startDate = "2026-01-01T00:00:00Z"``, validated at compile time) and
``java.time.Duration`` /
-``java.time.OffsetDateTime`` values in ``config`` calls. An unknown key or a
mismatched value type
-fails the build for an annotation, and the ``config`` call itself for an
object.
+**By position.** A ``@Airflow.TaskHandler`` method receives the arguments in
order. ``Client`` and ``Context``
+parameters, which the SDK injects, do not count:
-A Dag with a cron ``schedule`` runs in the time zone of its ``startDate``.
With no ``startDate`` it
-is scheduled in UTC, so set ``startDate`` to pin the zone.
+.. code-block:: python
-.. _java-sdk/task-state-store:
+ @task.stub(queue="java")
+ def score(rows, threshold): ...
-Task state store
-~~~~~~~~~~~~~~~~
-``client.getTaskStateStore()`` gives a task key-value state that is scoped to
the task instance and
-survives retry attempts within the same Dag run (see
:doc:`/core-concepts/task-state-store`). Use it to
-remember things like an external job ID so a retried task can resume instead
of starting over:
+ score(load_rows(), 0.75)
.. code-block:: java
- @Airflow.Task(id = "submit")
- public void submit(Client client) throws Exception {
- var store = client.getTaskStateStore();
- var jobId = (String) store.get("job_id");
- if (jobId == null) {
- jobId = submitJob();
- store.set("job_id", jobId, Duration.ofHours(6));
- }
- waitForJob(jobId);
- store.delete("job_id");
+ @Airflow.TaskHandler(dag = "etl", task = "score")
+ public long score(Client client, long rows, double threshold) {
+ // rows <- the value load_rows returned
+ // threshold <- 0.75
}
-``get`` returns ``null`` when the key is not set. ``set`` stores any
JSON-serializable value. Pass a positive
-``java.time.Duration`` to expire the key after that long,
``TaskStateStore.NEVER_EXPIRE`` for a key that
-garbage collection skips, or omit the retention to use the deployment's
``[state_store] default_retention_days``
-(0 means never expire). The coordinator passes that setting to the JVM as
-``AIRFLOW__STATE_STORE__DEFAULT_RETENTION_DAYS``. A zero or negative retention
is rejected. ``delete`` removes
-one key and ``clear`` removes every key for the task instance. The Java SDK
does not use a
-``[workers] state_store_backend``: values always go to the metadata database
as-is, so keys written by Python
-tasks through a custom backend are returned to Java as the raw reference
marker rather than the stored value.
+**By name.** A class implementing ``TaskInput`` receives each argument in the
public field of the same name,
+ignoring case and underscores, so ``region_code`` reaches ``regionCode``.
``@ArgName`` maps a field to an
+argument whose name cannot match on its own:
-.. _java-sdk/native-dag-parsing:
+.. code-block:: python
-Parsing native Java Dags
-~~~~~~~~~~~~~~~~~~~~~~~~
+ score(region_code="emea", threshold=load_threshold())
-To have Airflow parse the Dags a bundle JAR declares,
-put the JAR in a Dag bundle and configure a
:class:`~airflow.sdk.coordinators.java.JavaCoordinator`.
-The Dag processor runs the JAR's main class to list its Dags, so it needs a
Java executable, as the workers do:
+.. code-block:: java
-.. code-block:: ini
+ public static class ScoreInput implements TaskInput {
+ @ArgName("region_code")
+ public String region;
- [sdk]
- coordinators = {
- "java-native": {
- "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
- "kwargs": {"java_executable": "/usr/lib/jvm/java-17-openjdk/bin/java"}
- }
+ public double threshold;
}
- queue_to_coordinator = {"java-native": "java-native"}
-
-Once a ``JavaCoordinator`` is configured, the Dag processor parses the
executable JARs of every Dag bundle,
-so it needs this ``[sdk]`` configuration and a JRE. With one
``JavaCoordinator``, it parses them all.
-With several, map each Dag bundle that holds native Java Dags to one of them
in ``[sdk] dag_bundle_to_coordinator``.
-A JAR in a bundle that has no entry, or an entry that names no
``JavaCoordinator``, fails to parse with an import error:
-.. code-block:: ini
+ @Airflow.TaskHandler(dag = "etl", task = "score")
+ public long score(Client client, ScoreInput input) { ... }
- [sdk]
- dag_bundle_to_coordinator = {"dags-folder": "java-native"}
-
-A Dag bundle that holds only the JARs that Python Dags' tasks run should list
``*`` in its ``.airflowignore``
-either way: with one ``JavaCoordinator`` its JARs are still parsed on every
loop, and a JAR built with a Java
-SDK older than schema ``2026-10-30``, every released one today, fails to parse
as an import error. With
-several Java coordinators, an unmapped bundle's JARs fail to parse too.
-
-* Every JAR in the bundle whose manifest sets ``Main-Class`` is parsed. Each
Dag its main class
- declares, through ``Bundle.register`` of a ``DagDef`` or an ``@Airflow.Dag``
class, is stored with
- that JAR as its file. Task handlers for a Python Dag are not Dags. A JAR
without ``Main-Class`` is
- skipped, but many dependency JARs set one (the PostgreSQL JDBC driver and H2
do, for example), so a
- thin bundle should set ``main_class`` or list its dependency JARs in
``.airflowignore``.
-* Parsing needs a JAR built with a Java SDK whose supervisor schema version
(the
- ``Airflow-Supervisor-Schema-Version`` manifest attribute) is ``2026-10-30``
or later. An older JAR
- fails to parse, so list it in ``.airflowignore``; its tasks still find it,
because tasks do not read
- ``.airflowignore``.
-* Do not declare a Dag in Java that a Python file in the same bundle also
defines.
-* Keep one executable JAR per bundle, or set ``main_class``, so that only JARs
with that
- ``Main-Class`` are parsed. List JARs that should not be parsed in
``.airflowignore``.
-* Two JARs in the bundle that set the same ``Main-Class`` fail to parse,
because the JVM would load
- the classes of only one of them. Keep one in the bundle.
-* Set ``queue`` on every task, with ``@Airflow.Task(queue = "java-native")`` or
- ``TaskDef.config("queue", "java-native")``, so it runs on the coordinator's
queue. There is no
- Dag-level queue yet.
-* A task runs the JAR of its Dag on the ``JavaCoordinator`` that its queue
routes to,
- which need not be the one that parsed the JAR. For example, a queue can
route to a coordinator that uses another JDK.
- A task whose queue routes to another kind of coordinator fails without
retries. A task whose JAR is missing,
- or that the coordinator cannot run (for example, because ``main_class`` does
not match the JAR's ``Main-Class``),
- fails and retries while it has retries left.
-* The Code view shows the source of the JAR's main class, which the Gradle
plugin packs into the JAR.
-* Cluster policies (``dag_policy``, ``task_policy``) are not applied to a
native Java Dag.
-* ``airflow dags reserialize`` does not store the Dags of a JAR, which only
the Dag processor stores.
- ``airflow dags test``, ``tasks test`` and ``tasks render`` refuse a native
Java Dag.
- ``airflow tasks list`` lists its tasks by running the JAR's main class, so
it needs a JRE.
+With the interface, implement ``InputTask<I>`` instead of ``Task`` to receive
a ``TaskInput``:
-.. _java-sdk/logging:
+.. code-block:: java
-Logging
--------
+ public class ScoreTask implements InputTask<ScoreInput> {
+ @Override
+ public void execute(Context context, Client client, ScoreInput input) {
... }
+ }
-Task code can emit log records through any common Java logging framework. The
SDK ships optional
-integration libraries that forward those records to Airflow's task log store,
where they appear
-alongside the standard task output in the Airflow UI.
+The two forms handle a mismatch differently:
-Declare a logger as a static field on the task class, using the class's own
type as the name. This
-is the conventional pattern regardless of which logging framework you choose:
+.. list-table::
+ :header-rows: 1
+ :widths: 30 35 35
-.. code-block:: java
+ * - Case
+ - By position (method parameters)
+ - By name (``TaskInput`` fields)
+ * - More arguments than the handler declares
+ - The task fails.
+ - A warning is logged, and the extra arguments are ignored.
+ * - Fewer arguments than the handler declares
+ - The task fails.
+ - A warning is logged, and the field keeps its Java default.
+ * - An argument resolves to ``null``
+ - A boxed parameter (``Long``, ``Double``) receives ``null``; a primitive
one fails the task with
+ ``MissingXComException``.
+ - Same as by position.
+ * - Available in
+ - ``@Airflow.TaskHandler`` methods.
+ - ``@Airflow.TaskHandler`` methods and ``InputTask``.
+
+An argument the Python call leaves at its default does not count towards the
arguments by position, so the
+method may leave it out. A method declares flat parameters or one
``TaskInput``, not both, and two fields whose
+names match the same argument fail the build.
- private static final System.Logger log =
- System.getLogger(SalesPipeline.class.getName());
+.. note::
- @Airflow.TaskHandler(dag = "etl", task = "extract")
- public long extract(Client client) {
- log.log(System.Logger.Level.INFO, "Starting extraction");
- return recordCount;
- }
+ A dependency drawn with ``>>`` in the Python Dag orders the tasks and passes
nothing. To read a value the call
+ did not pass, use ``client.getXCom``.
-The Gradle snippets below show the dependency declarations; all Airflow
artifact versions are managed
-by ``airflow-sdk-bom``. Maven users apply the same artifact IDs following the
pattern in
-:ref:`java-sdk/build/maven`.
+Writing tasks
+-------------
-.. _java-sdk/logging/jpl:
+A task is an ordinary Java method or class. It works the same in both ways of
using the SDK.
-``System.Logger`` (Java Platform Logging)
-~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+A ``@Airflow.Task`` or ``@Airflow.TaskHandler`` method can declare a
``Client`` and a ``Context`` parameter in any
+position, and the SDK passes them in. A ``Task`` class receives both in
``execute``. ``Context`` describes the
+running task, such as ``context.ti.taskId``, ``context.ti.tryNumber`` and
``context.dagRun.logicalDate``.
-Java 9's new logging facade ``java.lang.System.Logger`` (JEP 264), commonly
abbreviated *JPL*, can be
-used by libraries without pulling in any third-party API. The
``airflow-sdk-jpl`` artifact registers an
-``AirflowSystemLoggerFinder`` via ``ServiceLoader``, which routes all
``System.Logger`` calls directly
-to Airflow's task log store.
+A value a method returns, or a ``Task`` class passes to ``client.setXCom``,
becomes the task's ``return_value``
+XCom, and is what a downstream task receives. An uncaught exception fails the
task, and retries it if the task
+has retries.
-.. code-block:: groovy
+The ``Client``
+~~~~~~~~~~~~~~
- implementation("org.apache.airflow:airflow-sdk-jpl:${version}")
+.. list-table::
+ :header-rows: 1
+ :widths: 40 60
-No configuration file or startup call is required. The ``ServiceLoader``
mechanism discovers the
-provider automatically as long as the JAR is on the classpath.
+ * - Method
+ - Description
+ * - ``getVariable(key)``
+ - The Variable's value, or ``null`` when it does not exist.
+ * - ``setVariable(key, value, description?)``
+ - Stores a Variable, replacing any existing value. Values are strings, so
serialize structured data, for
+ example to JSON.
+ * - ``deleteVariable(key)``
+ - Removes a Variable.
+ * - ``getConnection(connId)``
+ - A ``Connection`` with ``id``, ``type``, ``host``, ``schema``,
``login``, ``password``, ``port`` and
+ ``extra``. Throws when the connection does not exist.
+ * - ``getXCom(taskId)``
+ - The ``return_value`` XCom of a task in the same Dag run, or ``null``
when it does not exist. Overloads
+ take a ``key``, ``dagId``, ``runId`` and ``mapIndex``. See
:ref:`java-sdk/types`.
Review Comment:
Also `includePriorDates`? I’d rather this to not mention the arguments at
all and just tell the reader to refer to the API documentation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
