uranusjr commented on code in PR #74701:
URL: https://github.com/apache/airflow/pull/74701#discussion_r4335359714
##########
airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst:
##########
@@ -1112,296 +824,230 @@ represented as Java objects when read back via
``getXCom``.
- object
- ``Map<String, Object>``
-.. note::
-
- Avoid ``char`` and ``Character``. JSON has no single-character type, so
the value arrives
- as a string or a number and the SDK narrows it: a one-character string or
an integer code
- point binds, an empty string binds as ``null``, and anything longer fails
at runtime with
- an ``IllegalArgumentException``. Whether a value binds therefore depends
on its length
- rather than on the stub signature, which no compile-time check can catch.
Use ``String``
- instead.
-
-.. note::
-
- A data parameter whose binding resolves to a value that was never pushed
receives
- ``null``. A boxed parameter (``Integer``, ``Long``, ``Boolean``, …)
receives ``null``
- safely, but a primitive parameter (``int``, ``long``, ``boolean``, …)
cannot represent
- ``null`` and the task fails with ``MissingXComException``. Declare the
parameter with a
- boxed type when the upstream XCom may be absent.
-
-Variables
----------
-
-``Client`` reads, writes, and deletes Airflow Variables. Values are stored as
strings. Serialize
-structured data (for example to JSON) before storing it.
-
-.. code-block:: java
-
- var threshold = (String) client.getVariable("process_threshold");
- client.setVariable("process_threshold", "42", "Rows above this count take
the slow path");
- client.deleteVariable("legacy_threshold");
-
-.. note::
-
- A value supplied by a secrets backend (for example an ``AIRFLOW_VAR_*``
environment variable) still
- takes precedence over the stored value when the Variable is read back.
Calling ``setVariable``
- without a description clears any existing description.
+A typed parameter receives the type it declares, so a ``double`` parameter
accepts an integer and
+``List<Double>`` holds ``Double`` elements. Use ``String`` rather than
``char``: a ``char`` parameter fails on a string
+longer than one character.
.. _java-sdk/build:
-Building and packaging
------------------------
-
-The Java SDK is distributed as a JAR. The sections below show how to build a
bundle with Gradle or Maven.
+Building a bundle
+-----------------
.. _java-sdk/build/gradle:
Gradle
~~~~~~
-Apply the Airflow SDK Gradle plugin in your ``build.gradle``:
-
-.. code-block:: groovy
-
- plugins {
- id("org.apache.airflow.sdk") version "${version}"
- }
-
- dependencies {
-
annotationProcessor("org.apache.airflow:airflow-sdk-processor:${version}")
- implementation("org.apache.airflow:airflow-sdk:${version}")
- }
-
- airflowBundle {
- mainClass = "com.example.Main" // Point to your main class instead.
- }
-
-Then run:
+The ``org.apache.airflow.sdk`` Gradle plugin adds a ``bundle`` task,
configured by ``airflowBundle``, as in the
+:ref:`quick start <java-sdk/quick-start>`:
.. code-block:: bash
./gradlew bundle
-The ``build/bundle/`` directory contains all required JAR(s). Copy or mount it
into the Dag bundle named by
-``task_handler_bundle_name`` in the coordinator configuration.
-:class:`~airflow.sdk.coordinators.java.JavaCoordinator` scans that Dag bundle
recursively and builds the
-classpath automatically.
-
-The plugin also packs the source file of ``mainClass``, and of each class that
declares a Dag in Java,
-into the bundle JAR, so the Airflow UI can show the source of a native Java
Dag (see
-:ref:`java-sdk/native-dag-parsing`). To find those classes, the plugin runs
``mainClass`` once at build
-time. If that run fails, the build logs a warning and packs only the
``mainClass`` source.
+It writes the bundle to ``build/bundle/``, which is all you deploy. The plugin
also packs the source of each
+class that declares a Dag, so the Airflow UI's **Code** view shows it.
-A Dag is recorded against the class that constructed it. A Dag built by a
factory therefore maps to the
-factory's class, and one built by a factory in a dependency JAR has no source
in the project and falls
-back to the ``mainClass`` source.
-
-.. note::
-
- You only need the ``annotationProcessor`` entry if you use the
annotation-based API. It is not needed for
- the interface-based API.
-
-.. note::
-
- The plugin generates a fat JAR with the `Shadow
<https://gradleup.com/shadow/>`__ plugin by default. This is
- generally a good idea since you only deploy one JAR file to avoid dependency
issues between projects. If this
- does not suit you, set ``fatJar = false`` in ``airflowBundle`` to produce
thin JARs instead. The rest of the
- process stays the same, but you will need to put all dependency JARs in the
same Dag bundle.
+* By default the bundle is one fat JAR with all dependencies. Set ``fatJar =
false`` in ``airflowBundle`` for a
+ thin JAR, and deploy every dependency JAR in ``build/bundle/`` with it.
+* ``annotationProcessor`` is only needed for the annotation style.
.. _java-sdk/build/maven:
Maven
~~~~~
-Import the ``airflow-sdk-bom`` Bill of Materials so that artifact versions and
the
-``${airflow.supervisor.schema.version}`` property are managed in one place:
+Import ``airflow-sdk-bom`` to manage the SDK's artifact versions, and add the
SDK and, for the annotation
+style, the annotation processor:
.. code-block:: xml
<dependencyManagement>
- <dependencies>
- <dependency>
- <groupId>org.apache.airflow</groupId>
- <artifactId>airflow-sdk-bom</artifactId>
- <version>${version}</version>
- <type>pom</type>
- <scope>import</scope>
- </dependency>
- </dependencies>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.airflow</groupId>
+ <artifactId>airflow-sdk-bom</artifactId>
+ <version>1.0.0</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
+ </dependencies>
</dependencyManagement>
-Add the SDK as a dependency (version is managed by the BOM):
-
-.. code-block:: xml
-
<dependencies>
- <dependency>
- <groupId>org.apache.airflow</groupId>
- <artifactId>airflow-sdk</artifactId>
- </dependency>
+ <dependency>
+ <groupId>org.apache.airflow</groupId>
+ <artifactId>airflow-sdk</artifactId>
+ </dependency>
</dependencies>
-Wire the annotation processor through ``maven-compiler-plugin`` so it stays
off the runtime classpath:
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <annotationProcessorPaths>
+ <path>
+ <groupId>org.apache.airflow</groupId>
+ <artifactId>airflow-sdk-processor</artifactId>
+ <version>1.0.0</version>
+ </path>
+ </annotationProcessorPaths>
+ </configuration>
+ </plugin>
+
+Build a fat JAR with ``maven-shade-plugin``, setting the main class and the
supervisor schema version the BOM
+provides:
.. code-block:: xml
<plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <configuration>
- <annotationProcessorPaths>
- <path>
- <groupId>org.apache.airflow</groupId>
- <artifactId>airflow-sdk-processor</artifactId>
- <version>${version}</version>
- </path>
- </annotationProcessorPaths>
- </configuration>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals><goal>shade</goal></goals>
+ <configuration>
+ <transformers>
+ <transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>com.mycompany.airflow.hello.Main</mainClass>
+ <manifestEntries>
+
<Airflow-Supervisor-Schema-Version>${airflow.supervisor.schema.version}</Airflow-Supervisor-Schema-Version>
+ </manifestEntries>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
</plugin>
-**Option 1 (recommended): fat JAR**
+``mvn package`` writes the JAR to ``target/``. For thin JARs, set
``Main-Class`` with ``maven-jar-plugin`` and
+copy the runtime dependencies next to the JAR with
``maven-dependency-plugin``; the schema version is then read
+from the ``airflow-sdk`` JAR. Maven does not check that the main class exists,
so a wrong one fails at runtime.
-Use ``maven-shade-plugin`` to bundle your code and all dependencies into a
single JAR. This is the
-simplest deployment: one file, no dependency management at runtime.
+To show a Dag's source in the **Code** view, pack the main class's source
under ``META-INF/airflow/sources/``,
+add ``src/main/resources/META-INF/airflow/sources.json`` naming it, and point
the
+``Airflow-Java-SDK-Sources`` manifest entry at that file:
+
+.. code-block:: json
+
+ {"entrypoint_path": "com/mycompany/airflow/hello/Main.java"}
.. code-block:: xml
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <version>3.6.0</version>
- <executions>
- <execution>
- <phase>package</phase>
- <goals><goal>shade</goal></goals>
- <configuration>
- <transformers>
- <transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- <!-- Replace with the class that provides main. -->
- <mainClass>com.example.Main</mainClass>
- <manifestEntries>
- <!-- Resolved from the BOM; do not hard-code
this value. -->
-
<Airflow-Supervisor-Schema-Version>${airflow.supervisor.schema.version}</Airflow-Supervisor-Schema-Version>
- </manifestEntries>
- </transformer>
- </transformers>
- </configuration>
- </execution>
- </executions>
- </plugin>
+ <build>
+ <resources>
+ <resource>
+ <directory>src/main/resources</directory>
+ </resource>
+ <resource>
+ <directory>src/main/java/com/mycompany/airflow/hello</directory>
+ <includes><include>Main.java</include></includes>
+
<targetPath>META-INF/airflow/sources/com/mycompany/airflow/hello</targetPath>
+ </resource>
+ </resources>
+ </build>
-Then run:
+
<Airflow-Java-SDK-Sources>META-INF/airflow/sources.json</Airflow-Java-SDK-Sources>
-.. code-block:: bash
+Deploying
+---------
- mvn package
+Where the bundle goes depends on what it contains.
-The fat JAR is written to ``target/<artifactId>-<version>.jar``. Copy it into
the Dag bundle named by
-``task_handler_bundle_name`` in your coordinator.
+.. _java-sdk/native-dag-parsing:
-**Option 2: thin JAR with separate dependencies**
+Dag definition
+~~~~~~~~~~~~~~
-If a fat JAR does not suit your project, use ``maven-jar-plugin`` to set
``Main-Class`` on the regular
-JAR and ``maven-dependency-plugin`` to collect all runtime dependencies
alongside it. Note that
-``Airflow-Supervisor-Schema-Version`` does not need to be set here since
Airflow reads it directly from the
-``airflow-sdk`` JAR on the classpath.
+Put the bundle in a Dag bundle, such as the default ``dags-folder`` bundle.
The Dag processor runs each
+executable JAR it finds to read its Dags, as it reads any other Dag file, so
it needs a JRE and the same
+``[sdk]`` configuration as the workers:
-.. code-block:: xml
+.. code-block:: ini
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-jar-plugin</artifactId>
- <configuration>
- <archive>
- <manifestEntries>
- <Main-Class>com.example.Main</Main-Class>
- </manifestEntries>
- </archive>
- </configuration>
- </plugin>
+ [sdk]
+ coordinators = {"java": {"classpath":
"airflow.sdk.coordinators.java.JavaCoordinator"}}
+ queue_to_coordinator = {"java": "java"}
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-dependency-plugin</artifactId>
- <executions>
- <execution>
- <id>copy-dependencies</id>
- <phase>package</phase>
- <goals><goal>copy-dependencies</goal></goals>
- <configuration>
-
<outputDirectory>${project.build.directory}/bundle</outputDirectory>
- <includeScope>runtime</includeScope>
- </configuration>
- </execution>
- <execution>
- <id>copy-artifact</id>
- <phase>package</phase>
- <goals><goal>copy</goal></goals>
- <configuration>
- <artifactItems>
- <artifactItem>
- <groupId>${project.groupId}</groupId>
- <artifactId>${project.artifactId}</artifactId>
- <version>${project.version}</version>
-
<outputDirectory>${project.build.directory}/bundle</outputDirectory>
- </artifactItem>
- </artifactItems>
- </configuration>
- </execution>
- </executions>
- </plugin>
+With more than one ``JavaCoordinator``, map each Dag bundle that holds Java
Dags to one of them in
+``[sdk] dag_bundle_to_coordinator``, such as ``{"dags-folder": "java"}``, or
its JARs fail to parse.
-Then run:
+A task runs the JAR of its Dag from the same Dag bundle, at the version its
Dag run was created with, on the
+``JavaCoordinator`` its queue routes to.
-.. code-block:: bash
+* Every JAR whose manifest sets ``Main-Class`` is parsed. Many dependency JARs
set one, so keep one executable
+ JAR per Dag bundle, set the coordinator's ``main_class``, or list the other
JARs in ``.airflowignore``.
+* Two JARs that set the same ``Main-Class`` fail to parse.
+* A JAR built with an SDK older than ``1.0.0`` fails to parse. List it in
``.airflowignore``; tasks still
+ find it.
+* A bundle that fails to parse, or whose Dags have a cycle, shows up as an
import error, and the other Dags keep
+ working. Do not declare the same ``dag_id`` in a Python file of that Dag
bundle too.
- mvn package
+Python Dag with stub TaskHandler
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-``target/bundle/`` will contain the thin JAR and all runtime dependency JARs.
Copy or mount this
-directory into the Dag bundle named by ``task_handler_bundle_name``.
+By default, the coordinator loads the JARs that implement a stub task from the
stub task's own Dag bundle, at
+the version its Dag run was created with, so the JARs go next to the Python
Dag that declares the stub.
-.. note::
+To keep them in a Dag bundle of their own, name it with
``task_handler_bundle_name``. A task then uses the
+version that bundle is on when the task starts:
- You only need the ``annotationProcessorPaths`` entry if you use the
annotation-based API.
+.. code-block:: ini
-.. note::
+ [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"}
+ }
+ ]
+
+ [sdk]
+ coordinators = {
+ "java": {
+ "classpath": "airflow.sdk.coordinators.java.JavaCoordinator",
+ "kwargs": {"task_handler_bundle_name": "java-task-handlers"}
+ }
+ }
+ queue_to_coordinator = {"java": "java"}
- Unlike the Gradle plugin, Maven has no equivalent of the
``verifyBundleMainClass`` validation step.
- A wrong ``<mainClass>`` value will not be caught until runtime.
+Every JAR in that Dag bundle goes on one classpath, so all its handlers share
one set of dependencies. To isolate
+conflicting versions, give the handlers a second Dag bundle, served by a
second coordinator on its own queue.
-To show the source of a native Java Dag in the Airflow UI, pack the main
class's source file under
-``META-INF/airflow/sources/`` with an index that names it,
``src/main/resources/META-INF/airflow/sources.json``:
+The Dag processor also parses JARs that only implement stub tasks, and finds
no Dags in them. List such JARs in
+``.airflowignore`` to skip that; the coordinator still finds them. With its
own Dag bundle, ignore everything in
+it:
-.. code-block:: json
+.. code-block:: bash
- {"entrypoint_path": "com/example/Main.java"}
+ # [core] dag_ignore_file_syntax = glob (the default)
+ echo '*' > /opt/airflow/jars/sales-pipeline/.airflowignore
-.. code-block:: xml
+ # [core] dag_ignore_file_syntax = regexp
+ echo '.' > /opt/airflow/jars/sales-pipeline/.airflowignore
- <build>
- <resources>
- <!-- Declaring resources replaces the default, so keep it. -->
- <resource>
- <directory>src/main/resources</directory>
- </resource>
- <resource>
- <directory>src/main/java/com/example</directory>
- <includes><include>Main.java</include></includes>
- <targetPath>META-INF/airflow/sources/com/example</targetPath>
- </resource>
- </resources>
- </build>
+Where the configuration is needed
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-Then add
``<Airflow-Java-SDK-Sources>META-INF/airflow/sources.json</Airflow-Java-SDK-Sources>``
to the
-``manifestEntries`` of ``maven-shade-plugin`` or ``maven-jar-plugin`` shown
above.
+The coordinator runs where tasks run, so the ``[sdk]`` configuration and a JRE
are needed on the workers. With
+``CeleryExecutor``, set them on the Celery workers. With ``LocalExecutor``,
tasks run in the scheduler's process,
+so set them there. The Dag processor needs them too: it parses the JARs, and
checks the stub tasks of each
+Python Dag against the handlers the JARs register. The API server never needs
them.
+Register every Dag bundle a coordinator names in ``[dag_processor]
dag_bundle_config_list`` on every component.
Review Comment:
Is this correct? LocalExecutor does not run the task in the scheduler’s
process; it’s more like the scheduler creates a (sub)process with itself as the
host to run the task. The old text around line 280 was more accurate to me.
--
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]
