airflow-java-sdk

作者: astronomer

Guide for contributing to the Airflow Java SDK (AIP-108). Use this skill whenever a contributor is working in the `java-sdk/` directory or on the Java…

npx skills add https://github.com/astronomer/airflow --skill airflow-java-sdk

Airflow Java SDK contributor guide

The Java SDK lets Airflow tasks execute JVM code (Java, Kotlin, or any JVM language). You are helping a contributor work in one or both of these locations:

  • java-sdk/ — the JVM-side library (Kotlin source, published to Maven)
  • task-sdk/src/airflow/sdk/coordinators/java/ — the Python coordinator that launches the JVM subprocess

Read these two documents early in every session — they contain the authoritative reference material:

  • airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst — user-facing guide: annotation vs. interface API, XCom type mapping, Gradle/Maven steps, coordinator config.
  • java-sdk/README.md — contributor guide: repository layout, detailed execution walkthrough, Gradle + Breeze test commands, coding conventions, common tasks, and PR checklist.

SDK package architecture

The JVM-side library is split into two packages with distinct visibility rules:

  • org.apache.airflow.sdk — public, user-facing API. Classes here (e.g. Client, Bundle, BundleBuilder, Server) are stable contracts that DAG authors and task implementers import directly. Changes to this package are breaking changes.
  • org.apache.airflow.sdk.execution — internal implementation detail. Everything in this package (CoordinatorComm, LogSender, Log, Client in execution/, generated schema models, etc.) is not intended to be imported by users. It may change between releases without notice.

When reviewing or writing code, enforce this boundary: user task code and BundleBuilder subclasses must only import from org.apache.airflow.sdk; any import of org.apache.airflow.sdk.execution.* in user-facing API surface is a red flag.


Bundle composition and coordinator discovery

A bundle is a directory of JAR files (typically build/bundle/) placed on the coordinator's jars_root. The coordinator scans the directory at task-dispatch time to find:

  1. Main-Class (standard JAR manifest attribute) — the fully-qualified class name of the entry point that the coordinator invokes with java -classpath … <Main-Class> --comm … --logs …. This must be a class with a public static void main(String[] args) method; the Gradle plugin org.apache.airflow.sdk writes it automatically from airflowBundle { mainClass = "…" } and validates that the class exists and has the right signature at build time.

  2. Airflow-Supervisor-Schema-Version (Airflow-specific manifest attribute) — the wire protocol version the JVM side expects when talking to the Python supervisor. In fat-JAR mode (the default), the Gradle plugin reads this value from the airflow-sdk JAR in runtimeClasspath and copies it into the shadow JAR manifest. In thin-JAR mode (fatJar = false), the value stays in the airflow-sdk JAR deployed alongside the bundle JAR.

The Python coordinator (JavaCoordinator) scans every JAR under jars_root with _JarInfo.find(), reads META-INF/MANIFEST.MF out of each ZIP, and collects Main-Class and Airflow-Supervisor-Schema-Version from whichever JARs carry them. The resolved schema version is then passed as the schema_version return value from _build_execute_task_command, which the base SubprocessCoordinator uses to negotiate the supervisor wire protocol.

If main_class is set explicitly on the JavaCoordinator instance (via [sdk] coordinators kwargs), the scan uses it as a filter; otherwise the first JAR with a Main-Class attribute wins. Either way, Airflow-Supervisor-Schema-Version must be present in at least one JAR in jars_root or startup fails.


Key files to know

FilePurpose
java-sdk/sdk/.../Client.ktPublic API (Variables, Connections, XCom)
java-sdk/sdk/.../execution/Client.ktSupervisor wire calls
java-sdk/sdk/.../execution/Comm.kt4-byte-prefix MessagePack framing
java-sdk/sdk/.../Server.ktEntry-point; drives the execution loop
java-sdk/processor/.../BuilderProcessor.ktKapt annotation processor
java-sdk/plugin/.../AirflowSdkPlugin.ktGradle bundle plugin
task-sdk/.../coordinators/java/coordinator.pyPython side — spawns the JVM
task-sdk/.../schema/schema.jsonWire protocol definition (both sides)

Running tests

Always use ./gradlew from inside java-sdk/; never run Gradle via apt's gradle. See java-sdk/README.md#testing for the full list of Gradle commands.

For the Python coordinator, use Breeze (never pytest directly on the host):

breeze testing task-sdk-tests -- task_sdk/coordinators/java

End-to-end test suite:

E2E_TEST_MODE=java_sdk uv run --project airflow-e2e-tests pytest \
    tests/airflow_e2e_tests/java_sdk_tests/ -xvs

Updating the Python coordinator

coordinator.py extends SubprocessCoordinator. The only method subclasses must implement is _build_execute_task_command, which returns (argv, schema_version). Look at the existing implementation for how jars_root, java_executable, jvm_args, and main_class are assembled into the command. Do not reach into the JVM process from Python beyond what this method provides.


Upgrading Supervisor Schema client

When upgrading to a newer Supervisor Schema version:

  • Regenerate models with ./gradlew generateJsonSchema2Pojo
  • Modify execution/Client.kt to handle changes

The java-sdk/README.md#contributing section walks through the full "adding a new Client method" sequence step by step.

来自 astronomer 的更多技能

airflow
astronomer
查询、管理和排查Apache Airflow的DAG、运行记录、任务及系统配置。支持30多种命令,涵盖DAG检查、运行管理、任务日志、配置查询及直接REST API访问。通过持久化配置管理多个Airflow实例;自动发现本地和Astro部署。同步(等待完成)或异步触发DAG运行,诊断故障,清除运行记录以重试,并通过重试/映射索引过滤访问任务日志。输出...
official
airflow-hitl
astronomer
在Airflow DAG中使用可延迟操作符实现人工审批关卡、表单输入和分支。四种操作符类型:用于批准/拒绝决策的ApprovalOperator、带表单的多选项选择HITLOperator、人工驱动的任务路由HITLBranchOperator,以及表单数据收集HITLEntryOperator。所有操作符均为可延迟设计,在通过Airflow UI的"必需操作"标签页或REST API等待人工响应时释放工作槽位。支持包括自定义在内的可选功能...
official
airflow-state-store
astronomer
Persists task and asset state across retries and DAG runs using Airflow 3.3's AIP-103 key/value stores (`task_state_store`, `asset_state_store`) and the…
official
analyzing-data
astronomer
查询数据仓库,利用缓存的模式和概念映射来回答业务问题。支持对重复问题类型进行模式查找和缓存,并通过记录结果来改进后续查询。包含概念到表的映射缓存,以及通过INFORMATION_SCHEMA或代码库grep进行表结构发现。提供run_sql()和run_sql_pandas()内核函数,返回Polars或Pandas DataFrame用于分析。提供CLI命令用于管理概念、模式和表缓存,以及...
official
annotating-task-lineage
astronomer
使用入口和出口为Airflow任务标注数据血缘。支持使用OpenLineage Dataset对象、Airflow Assets和Airflow Datasets定义跨数据库、数据仓库及云存储的输入输出。当运算符缺少内置OpenLineage提取器时作为备用方案;遵循四级优先级系统,其中自定义提取器和OpenLineage方法优先。包含针对Snowflake、BigQuery、S3和PostgreSQL的数据集命名辅助工具,以确保一致性...
official
authoring-dags
astronomer
创建Apache Airflow DAG的引导式工作流,集成验证与测试。采用六阶段结构化方法:发现环境与现有模式、规划DAG结构、遵循最佳实践实现、通过af CLI命令验证、经用户同意测试、迭代修复。用于发现(af config connections、af config providers、af dags list)和验证(af dags errors、af dags get、af dags explore)的CLI命令可提供DAG的即时反馈...
official
authoring-go-sdk-tasks
astronomer
Writes Airflow task logic in Go using the Airflow Go SDK. Use when the user wants to implement Airflow tasks in Go, asks about `BundleProvider`/`RegisterDags`,…
official
authoring-java-sdk-tasks
astronomer
使用 Airflow Java SDK 编写 Java、Kotlin 或任何 JVM 语言的 Airflow 任务逻辑。当用户想要在 Java/JVM 中实现 Airflow 任务时使用,询问……
official