尧图网页设计 
Apache Airflow Java SDK 开发与运行指南从 JVM 工作流 Bundle 到发布全流程【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowApache Airflow Java SDK 是 Airflow 为 JVM 生态提供的语言 SDK允许开发者用 Java 或其他 JVM 兼容语言编写工作流任务并以可被 Airflow 消费的 Bundle 形式交付执行。本文以java-sdk/README.md为骨架结合仓库中的源码、示例与配置系统讲解 SDK 的模块构成、构建与依赖校验、示例运行方式、任务编写 API、执行期通信协议、兼容性矩阵以及从本地发布到 Maven Central 的完整发布与投票流程。读完本文你将掌握如何构建 SDK、用 Java 编写并打包工作流、将其接入 Airflow 的 Java 队列以及如何独立核验一个 Java SDK 发布候选。一、SDK 定位与运行环境Java SDK 是一个面向JVM的 Apache Airflow 开发套件。你可以使用任意 JVM 兼容语言编写工作流 Bundle再由 Airflow 消费执行结果。SDK 本身与执行期逻辑使用Kotlin实现但公开 API 面向 Java仓库中捆绑了一个使用 Java 调用 SDK 的端到端示例并在scala_spark_example/中提供了 Scala Spark 的使用示例见 ScalaSparkExample.scala。运行环境要求SDK 运行时要求Java 11 或更高版本可选组件与开发工具可能有更高要求具体见 airflow-jvm-conventions.gradle.kts 中定义的 toolchain服务端侧要求Airflow 3.3 及以上版本且 supervisor 通信 schema 版本为2026-06-16见 capabilities.yaml 与gradle.properties中的airflowSupervisorSchemaVersion。更多 Java SDK 的使用细节可参见 Airflow 官方文档Authoring and Scheduling下的 Java SDK 章节。二、仓库布局一个 JVM 多模块 Gradle 工程java-sdk/是一个多模块 Gradle 工程各模块职责清晰源码结构见 java-sdk/settings.gradle.kts模块职责sdk/核心库公开 APIorg.apache.airflow.sdk与内部执行层org.apache.airflow.sdk.execution含Server、Client、Bundle、DagDef、Builder等核心类型processor/注解处理器为被Builder.Dag标注的类生成*Builder类BuilderProcessor.kt基于 kaptplugin/Gradle 插件org.apache.airflow.sdk提供bundle打包任务、Manifest 属性注入与verifyBundleMainClass校验bom/Bill of Materials POM便于消费者以统一版本导入全部 SDK 构件slf4j/SLF4J 日志 Provider将 SLF4J 调用路由到 Airflow 日志存储jul/java.util.loggingHandler将 JUL 记录路由到 Airflow 日志存储对应AirflowJulHandler.ktjpl/Java Platform Logging ProviderSystem.LoggerJEP 264路由 JPL 调用对应AirflowSystemLoggerFinder.ktlog4j2/Log4j 2 Appender路由 Log4j 2 事件到 Airflow 日志存储对应AirflowLog4jAppender.javaexample/端到端示例 Bundle注解 API 接口 APIJava 源码buildSrc/共享 Gradle 约定插件Java 版本、lint、格式化等Python 侧的 JVM 启动协调器不在本目录内位于task-sdk/src/airflow/sdk/coordinators/java/JavaCoordinatorSubprocessCoordinator子类task-sdk/tests/coordinators/java/Python 侧单元与集成测试。三、构建 SDK 与依赖校验构建整个 SDK 只需在java-sdk/目录下执行./gradlew build依赖变更审查gradle/verification-metadata.xml为依赖、插件及其元数据固定了 SHA-256 校验和。只要该文件存在Gradle 会自动启用依赖验证并默认采用strict模式因此普通的./gradlew build就已经在执行校验。strict模式会在以下两种情况下失败校验和不匹配某个构件在元数据中完全没有条目——这通常是升级依赖时最常见的情况意味着构建解析到了元数据未描述的内容。仓库层面还有两条约定仓库集中在settings.gradle.kts中声明且**动态版本dynamic与变化版本changing**对项目依赖配置是被禁止的插件 marker 与 detached 配置除外需要手工固定版本buildSrc/自行声明仓库但其依赖同样被该元数据覆盖。更新依赖或插件时应从可信网络重新生成元数据。注意任务列表必须覆盖 CI 运行的全部内容因为只有被调用任务实际解析到的依赖才会被记录./gradlew --write-verification-metadata sha256 --refresh-dependencies \ build \ :sdk:dokkaGeneratePublicationHtml :sdk:dokkaGeneratePublicationJavadoc \ sourceTarball checksumSourceTarball \ publishToMavenLocal -PskipSigningtrue几点说明不带-PskipSigningtrue时签名任务会失败但 Gradle 仍会从部分运行中写出元数据重新生成只会追加版本升级后过期的条目需要手工删除生成文件只记录仓库当时提供的内容并不代表这些字节可信务必把新坐标与校验和与依赖官方发布信息交叉核对生成元数据不覆盖example/、scala_spark_example/、kubernetes-tests/lang_sdk/java_example/的构建、foojay resolver 自动供应的 JDK以及:sdk:syncSupervisorSchema拉取的 Supervisor Schema。构建文档./gradlew dokkaGenerate该命令使用 Dokka 构建 Java SDK 文档同时生成 HTML 表示与 Javadoc含:sdk:dokkaGeneratePublicationHtml与:sdk:dokkaGeneratePublicationJavadoc两个变体。四、端到端运行示例要让示例真正跑起来需要完成 SDK 构件发布、Bundle 打包、Airflow 队列配置与连接/变量准备四个环节。1. 发布 SDK 到本地 Maven 仓库./gradlew publishToMavenLocal -PskipSigningtrue构建成功后~/.m2/repository/org/apache/airflow/下会出现airflow-sdk、airflow-sdk-processor、airflow-sdk-bom、airflow-sdk-gradle-plugin等目录。2. 打包示例 Bundle进入示例工程并执行 bundle 任务# 已进入 example 目录gradlew 在父目录 cd example ../gradlew bundleBundle 会输出到example/build/bundle。打包由 Gradle 插件AirflowSdkPlugin.kt负责包括 bundle 任务、Manifest 属性注入与verifyBundleMainClass校验。3. 放置 DAG 文件将带 stub 任务的 DAG 放到 Airflow 能找到的位置仓库提供了现成示例java_examples.py。4. 配置 Airflow 的 Java 队列确保 Airflow 任务 worker 所在环境能执行java命令然后配置[sdk]段的协调器把java队列的任务路由给 Java 执行export AIRFLOW__SDK__COORDINATORS{ java: { classpath: airflow.sdk.coordinators.java.JavaCoordinator, kwargs: {jars_root: [/opt/airflow/java-sdk/example/build/bundle]} } } export AIRFLOW__SDK__QUEUE_TO_COORDINATOR{java: java}classpath指向 Python 侧JavaCoordinator的完整导入路径kwargs.jars_root是扫描 JAR Bundle 的目录列表QUEUE_TO_COORDINATOR建立队列名 → 协调器名的映射将java队列路由到名为java的协调器条目。5. 准备连接与变量示例 DAG 依赖的 Connection 与 Variable 可通过环境变量注入export AIRFLOW_CONN_TEST_HTTP{ conn_type: http, login: user, password: pass, host: example.com, port: 1234, extra: {param1: val1, param2: val2} } export AIRFLOW_VAR_MY_VARIABLE123五、编写任务注解 API 与接口 APISDK 提供两套编写任务的 API注解驱动的声明式 API以及面向底层编排的接口 API。1. 注解 API推荐org.apache.airflow.sdk.Builder容器类Builder.kt定义了三枚注解由processor模块的BuilderProcessor.kt在编译期生成*Builder类注解目标参数说明Builder.Dag类id默认取类名、to生成的 Builder 类名默认类名 Builder标注一个 DAG 定义类处理器为其生成FooBuilder.build()返回装配好的DagDefBuilder.Task方法id默认取方法名将方法标注为任务定义Builder.XCom方法参数task默认取参数名将参数标记为来自指定任务的 XCom 输入仓库中的 AnnotationExample.java 展示了完整用法Builder.Dag(id java_annotation_example) public class AnnotationExample { Builder.Task(id extract) public long extractValue(Client client) throws InterruptedException { var pythonXcom client.getXCom(python_task_1); // 读取 Python 任务推送的 XCom var connection client.getConnection(test_http); // ... 业务逻辑 return new Date().getTime(); // 返回值自动作为 return_value XCom 推送 } Builder.Task(id transform) public long transformValue(Client client, Builder.XCom(task extract) long extracted) { var variable client.getVariable(my_variable); return new Date().getTime(); } }要点任务方法参数中的Client由运行时自动注入用Builder.XCom标注的参数会在任务执行前自动拉取上游任务同名或task指定的 XCom方法的返回值会自动以return_value为键推送为 XCom供下游任务消费常量XCOM_RETURN_KEY定义在 Client.kt若某参数声明为原始类型如long而对应 XCom 从未被推送会抛出MissingXComException——改用装箱类型如Long即可接收null见 Client.kt 的MissingXComException构造逻辑。示例中还演示了重试感知load任务在第一次执行context.ti.tryNumber 1时故意抛异常由于 Java SDK 会在ti_context.should_retry置位时返回RetryTask而非终态的FAILEDsupervisor 会把任务标记为UP_FOR_RETRY重试后任务正常完成。同时concurrent任务验证了单个 supervisor 通道可以承载跨线程并发客户端调用8 线程 × 32 次getConnection。2. 接口 API底层编排当需要底层控制时可直接使用 DagDef.kt 中的类型DagDefDAG 定义ID 只能包含 ASCII 字母数字、横线、点或下划线且在一个 Bundle 内唯一通过addTask(id, Class)链式注册任务TaskDef单个任务定义ID 实现Task接口的类类需有公共无参构造器Task接口实现execute(context, client)方法抛出的任何异常都会将任务实例标记为失败。var dag new DagDef(java_etl) .addTask(extract, Extract.class) .addTask(load, Load.class);3. Bundle 与进程入口Bundle.kt 定义Bundle为当前 JVM 进程可执行的全部DagDef的不可变快照DAG ID 重复会抛出IllegalArgumentExceptionBundleBuilder接口负责收集 DAG 并提供build()。进程入口通过Server启动见 ExampleBundleBuilder.javapublic class ExampleBundleBuilder implements BundleBuilder { Override public IterableDagDef getDags() { return List.of( InterfaceExampleBuilder.build(), AnnotationExampleBuilder.build(), XComCastingExampleBuilder.build()); } public static void main(String[] args) { var bundle new ExampleBundleBuilder().build(); Server.create(args).serve(bundle); } }ServerServer.kt是 SDK 的运行核心Server.create(args)解析两个由 Airflow 自动注入的命令行参数--comm host:port任务执行消息通道与--logs host:port日志转发通道无需手工构造serve(bundle)是阻塞入口对serveAsync的包装连接协调器后分发任务执行请求当协调器关闭连接通常在一次任务实例执行后时进程退出进程启动时会同时打开--comm与--logs两个 socket随后等待首帧消息若收到StartupDetails则按dag_idtask_id查表调用用户任务方法若收到ErrorResponse则抛出ApiError。4. 运行时 Client API任务方法中注入的ClientClient.kt封装了与 Airflow supervisor 的通信所有读写默认自动限定在当前 DAG run 与任务实例范围内方法作用getConnection(id)读取 Airflow 连接存储中的连接返回Connection含 id/type/host/schema/login/password/port/extra 字段连接不存在或调用失败抛ApiErrorgetVariable(key)读取 Airflow 变量未设置时返回nullgetXCom(key, dagId, taskId, runId, mapIndex, includePriorDates)读取其他任务推送的 XCommap_index为null时对 mapped 任务返回按 map index 升序聚合的集体结果列表setXCom(key, value)推送 XCom 供下游读取值必须可 JSON 序列化六、执行架构与线协议JavaCoordinator与 JVM 子进程的协作流程如下详见 README Contributing 一节与 coordinator.py启动 JVMAirflow supervisor 判定任务应在 Java 队列运行后调用JavaCoordinator.execute_task()Python。该协调器扫描jars_root构建 classpath并执行java -cp jars MainClass --commhost:port --logshost:port连接建立Server.kt启动后立即连接两个 socket下发任务supervisor 发送StartupDetailsMessagePack 消息JVM 读取后按dag_idtask_id找到匹配任务并调用用户任务方法运行期请求执行期间 JVM 通过 comm 通道向 supervisor 发送请求GetVariable、GetConnection、GetXCom、SetXCom 等并等待响应结束任务完成或抛异常时JVM 发送TaskState消息并关闭 socket进程随之退出。关键实现细节Comm.kt帧格式所有帧均为4 字节大端长度前缀 MessagePack 载荷请求/响应关联每个请求获得递增的id通过pending表ConcurrentHashMapInt, CompletableDeferredIncomingFrame与读循环协程配对响应communicateT()对ErrorResponse自动抛ApiError帧大小防护入站帧上限取协议规定的Frame.MAX_FRAME_LENGTH与堆内存 1/8MAX_HEAP_FRACTION_PER_FRAME 8随-Xmx自适应的较小值超限帧在分配内存前即被拒绝将不可恢复的 OOM 转化为可捕获的FrameProcessingException流式解码ChannelFrameInput以 64 KiB 分块喂给 MessageUnpacker避免超大帧一次性大分配长度前缀承诺的字节数多于实际载荷under-run会被视为流失步日志转发SDK 自身而非用户代码产生的日志消息通过--logssocket 转发由 supervisor 追加到 Airflow 日志存储。对应实现见execution/Logger.kt与各日志桥接模块SLF4J/JUL/JPL/Log4j 2。线协议由 task-sdk/src/airflow/sdk/execution_time/schema/schema.json 统一定义新增消息类型需要同时修改 Python 侧schema.json与 JVM 侧execution/Comm.ktexecution/Client.kt。关于协调器架构与各语言 SDK 共享的核心集成面可进一步阅读 airflow-core/adr/lang-sdk 下的架构决策记录ADR。从 Python 协调器源码看JavaCoordinator支持以下配置项见 coordinator.py 类 docstring 与字段定义参数默认值说明java_executablejavajava命令路径默认依赖$PATHjvm_args[]额外 JVM 参数如[-Xmx1024m]jars_root必填至少 1 个扫描 JAR Bundle 的目录列表main_class显式入口类未指定时扫描jars_root寻找带Main-Class元数据的可执行 JAR若存在多个可能不确定task_startup_timeout10 秒等待任务进程启动的最长时间JavaCoordinator._build_execute_task_command()会从 JAR 的META-INF/MANIFEST.MF中读取Main-Class与Airflow-Supervisor-Schema-Version两个条目后者用于核对线协议 schema 版本——Java SDK 打包的 JAR 会自动写入该元数据只要依赖 JAR 原样部署就无需额外处理若重新打包依赖则必须在一个 JAR 中复现该条目。七、兼容性矩阵README 中的兼容性矩阵由 capabilities.yaml 自动生成update-java-sdk-readme-matrixpre-commit 钩子负责重生成请勿手工编辑表格符合 contributing-docs/30_new_language_sdk.rst 定义的 Language SDK 一致性规范。当前 SDK 的能力概览对应 Airflow 3.3、supervisor schema2026-06-16TaskInstance 状态支持MUST3.3success、failed、up_for_retry通过RetryTask、removed未支持skipped、deferred、up_for_reschedule、awaiting_input运行时尚未发出对应消息。运行时能力支持MUST3.3mixed-lang-stub-targettask.stub、task-loggingSLF4J JPL 桥接到任务日志、xcom-read-write、connection-read、self-contained-bundleAirflow 元数据内嵌于 jar 构件未支持variable-read-write目前仅getVariable尚无写通道、retry-policy、task-state-store、asset-state-store、asset-event-emit、asset-event-read。Native-Dag 编写native-dag-authoring尚未实现因此task-args、dag-params、taskflow-dependencies、branching、dag-test、task-group、dynamic-task-mapping、asset-inlets-outlets、asset-scheduling、object-store等原生能力均标注为 n/a仅在 native-dag-authoring 支持后才适用。在 Airflow 侧混合语言场景通过task.stub声明 stub 任务并指定队列由 supervisor 把对应队列的任务路由给 Java 协调器执行Java 任务通过 XCom 与 Python 任务交换数据示例 DAG java_examples.py 中可见python_task_1/python_task_2与 Java 任务的 XCom 互通。八、发布流程SDK 通过 ASF Nexus staging 仓库发布到 Maven Central。凡发布到 Maven Central而非 Snapshots的版本均视为正式发布含 alpha、beta 等每次发布都需 PMC 投票仅-SNAPSHOT构建可免投票发布。前置条件具备访问 repository.apache.org 的 ASF committer 账号以及已加入项目 KEYS 文件并上传到公共密钥服务器的 GPG 密钥。1. 版本号编辑gradle.properties设置projectVersionVERSION或对单次命令用-PprojectVersionVERSION覆盖。版本字符串需符合 Maven 版本顺序规范例如 beta 1 写作1.0.0-beta1。main分支在发布之间应保持在-SNAPSHOT版本snapshot 排序在VERSION之后、GA 之前因此 beta 之后无需额外 bump。2. 打 RC 标签git tag -s java-sdk/VERSION-rcN -m Java SDK VERSION RC N git push upstream java-sdk/VERSION-rcNRC 编号保留在标签名中投票失败只需递增到下一 RC构件版本本身不带 RC 后缀。投票前需先推送标签便于评审者检出确切的待投票源码。3. 本地核验 POMrm -rf ~/.m2/repository/org/apache/airflow/ # 清空旧构件 ./gradlew publishToMavenLocal -PskipSigningtrue less ~/.m2/repository/org/apache/airflow/airflow-sdk/*/airflow-sdk-*.pom less ~/.m2/repository/org/apache/airflow/airflow-sdk-bom/*/*.pom less ~/.m2/repository/org/apache/airflow/airflow-sdk-processor/*/airflow-sdk-*.pom less ~/.m2/repository/org/apache/airflow/airflow-sdk-gradle-plugin/*/airflow-sdk-*.pom less ~/.m2/repository/org/apache/airflow/sdk/org.apache.airflow.sdk.gradle.plugin/*/*.pom核对各 POM 的坐标、描述、license、SCM 与 organization 字段。4. 对本地仓库做发布演练rm -rf /tmp/local-maven-repo ./gradlew publish -PmavenUrlfile:///tmp/local-maven-repo -PskipSigningtrue ls /tmp/local-maven-repo/org/apache/airflow/输出应与上一步~/.m2中的构件一致。本地演练不需要签名如需测试签名按下一节配置 GPG 私钥与口令并去掉-PskipSigningtrue。5. 发布到 ASF Nexus staging将凭据写入~/.gradle/gradle.properties避免进入 shell 历史mavenUsernameyour-asf-nexus-token-username mavenPasswordyour-asf-nexus-token-password signing.passwordyour-gpg-key-passphrase然后暂存并关闭发布根build.gradle.kts应用的 Gradle Nexus Publish Plugin 会把所有模块聚合进同一个staging 仓库无需逐模块协调./gradlew publishToApache closeApacheStagingRepository \ --no-configuration-cache \ -Psigning.key$(gpg --armor --export-secret-keys your-gpg-key-fingerprint)注意三点签名密钥通过命令行传入因其含换行符不适合放在 Gradle properties 文件也可用环境变量提供凭据ASF_NEXUS_USERNAME、ASF_NEXUS_PASSWORD、SIGNING_KEY、SIGNING_PASSWORD适合 CI项目全局启用 configuration cache但 staging 任务publishToApache、closeApacheStagingRepository、releaseApacheStagingRepository通过 Nexus REST API 工作且不兼容 configuration cache因此发布命令必须加--no-configuration-cache。发布后在 Nexus 的Staging Repositories中打开已关闭的仓库核验所有模块均包含 jar、-sources.jar、-javadoc.jar适用处、.pom与.asc签名并检查Updated by、Uploaded Date、Last Modified字段。6. 上传源码包投票正式针对的是已签名的源码包Maven 构件是 convenience binaries。sourceRelease任务基于已提交的java-sdk源码含LICENSE与NOTICE一次生成签名与校验和./gradlew sourceRelease -PgitRefjava-sdk/VERSION-rcN输出到build/distributions/的三个文件apache-airflow-java-sdk-VERSION-src.tar.gz apache-airflow-java-sdk-VERSION-src.tar.gz.asc apache-airflow-java-sdk-VERSION-src.tar.gz.sha512注意源码包刻意省略 Gradle wrapper 脚本gradlew、gradlew.bat与gradle/wrapper/gradle-wrapper.jar——ASF 源码发布不得包含编译产物见 LEGAL-570而缺少 jar 的脚本没有用处gradle/wrapper/gradle-wrapper.properties会保留以固定 Gradle 版本与发行版校验和供重新生成 wrapper 时核验。将三个文件复制到 ASF distdev仓库并提交svn checkout https://dist.apache.org/repos/dist/dev/airflow dist-dev-checkout cd dist-dev-checkout mkdir -p java-sdk/VERSION-rcN cp path-to/java-sdk/build/distributions/apache-airflow-java-sdk-VERSION-src.tar.gz* \ java-sdk/VERSION-rcN/ svn add --parents java-sdk/VERSION-rcN svn commit -m Add Apache Airflow Java SDK VERSION-rcN source release candidate7. 发起投票向devairflow.apache.org发送[VOTE]邮件链接 git 标签与提交、dist/dev中的源码包、已关闭的 Nexus staging 仓库及KEYS文件。投票至少开放 72 小时。README 提供了完整模板subject 为[VOTE] Release Apache Airflow Java SDK VERSION based on VERSION-rcN正文列出 Maven 构件清单、Git 信息、源码包与 convenience binaries URL、投票选项。发送前检查所有构件能通过 BOM 交叉解析、Changes since rc 与截止时间已填写、对渲染后的邮件执行grep email.txt确认无输出任何匹配都说明占位符未填完。8. 独立核验发布候选任何社区成员都应在投票前独立核验候选核验清单如下校验和sha512sum -c apache-airflow-java-sdk-VERSION-src.tar.gz.sha512签名下载并导入KEYS文件用gpg --verify验证.asc签名与 git 标签比对解包 tarball与标签的干净检出做diff -rq除.gitattributesexport-ignore排除的文件gradlew、gradlew.bat、gradle-wrapper.jar、scripts外应无差异解包后的顶层目录应为apache-airflow-java-sdk-version不带仅出现在压缩包文件名中的-src后缀无二进制文件ASF 源码发布不得含编译产物用file扫描非文本文件应无输出从源码构建用本地 Gradle 依gradle-wrapper.properties中的distributionUrl与distributionSha256Sum重新生成 wrapper再./gradlew buildstaged 二进制冒烟测试从一个临时工程指向 staging 仓库 URL声明org.apache.airflow:airflow-sdk-bom:VERSION依赖确认传递构件含airflow-sdk-jpl可解析、示例 Bundle 能构建对应脚本见 scripts/ci/smoke-test-staged-binaries.sh 与 scripts/ci/verify-source-release.sh。9. 投票成功后的收尾回复[RESULT][VOTE]统计然后releasestaging 仓库同步到 Maven Central 需数小时不重新构建或签名./gradlew releaseApacheStagingRepository --no-configuration-cache将源码包从dist/dev移至dist/releasesvn mv在同一被投票的提交上打最终版本标签java-sdk/VERSION保留 RC 标签以便追溯发布 GitHub release附带被投票且已签名的源码构件gh release create预发布版本加--prerelease核验标签用--verify-tag等待约 1 小时Maven Central 同步后发送纯文本[ANNOUNCE]邮件到usersairflow.apache.org抄送dev并在 ASF Committee Report Helper 中记录发布触发Publish Docs to S3workflow 发布 API 文档确认https://airflow.apache.org/docs/java-sdk/stable/可解析且/docs/java-sdk/重定向到它。若投票失败关闭投票、在 Nexus 中dropstaging 仓库、删除dist/dev候选、修复问题后切下一 RC...-rc2。发布版本号不变仅标签中的 RC 计数递增。九、测试与编码规范运行测试# 运行全部 JVM 测试 ./gradlew test # 运行指定测试类 ./gradlew :sdk:test --tests org.apache.airflow.sdk.execution.CommTestPython 协调器测试必须通过 Breeze 运行不要在宿主机直接跑 pytestbreeze testing task-sdk-tests -- task_sdk/coordinators/java端到端测试需要真实 Airflow 环境E2E_TEST_MODEjava_sdk uv run --project airflow-e2e-tests pytest \ tests/airflow_e2e_tests/java_sdk_tests/ -xvs编码规范所有 SDK 与 processor 源码均为KotlinJava 是公开 API 目标而非实现语言保持sdk/src/main/kotlin/公开 API 面不混入内部实现细节内部实现属于execution/子包注解处理器BuilderProcessor.kt使用 kapt新增注解时在Builder.kt定义、在BuilderProcessor.kt处理并在processor/src/test/kotlin/添加 golden-output 测试Python 协调器继承SubprocessCoordinator除_build_execute_task_command提供的接口外不要从 Python 深入 JVM 进程内部提交前运行./gradlew ktLintCheck spotlessCheck或ktLintFormat spotlessApply项目强制 Kotlin 与 Java 格式所有新文件必须带 Apache License 头。常见开发任务新增一个Client方法如新的 Airflow API 调用若消息类型是新的从schema.json重新生成 POJO在execution/Comm.kt或新文件添加 Kotlin 请求/响应数据类在公开Client.kt添加委托给execution/Client.kt做 wire 调用的方法在sdk/src/test/kotlin/.../ClientTest.kt编写 mock socket 层的单元测试若变更用户可见同步更新airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst。新增注解在Builder.kt定义注解接口 → 在BuilderProcessor.kt生成对应代码 → 在BuilderTest.kt添加期望生成输出的测试 → 更新java.rst的注解表。修复组帧或协议 bug聚焦execution/Comm.kt与execution/Frame.ktCommTest.kt覆盖编解码往返修复前先添加复现该 bug 的回归测试。PR 检查清单运行./gradlew build testJVM与对应 pytest 套件Python 协调器确认示例 Bundle 仍可编译按运行示例一节做到打包步骤若schema.json变更确认 JVM 与 Python 两侧都能处理新/改字段为每项行为变更添加或更新测试对task-sdk/的用户可见变更在airflow-core/newsfragments/下添加 newsfragment。十、小结Apache Airflow Java SDK 让 JVM 团队以惯用的 Java/Kotlin/Scala 方式编写 Airflow 任务通过Builder.Dag/Builder.Task/Builder.XCom注解声明式描述 DAG用 Gradle 插件一键打成自包含 Bundle由JavaCoordinator以子进程方式拉起 JVM 并通过 4 字节长度前缀 MessagePack 的 comm 通道与 supervisor 协作天然支持与 Python 任务的 XCom 互通和task.stub混合语言调度。其发布流程遵循 ASF 规范从本地 POM 核验、staging 发布、源码包签名、PMC 投票到最终同步 Maven Central 均有清晰的步骤与可执行的验证清单为在 Java 生态中落地 Airflow 提供了完整链路。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Apache Airflow Java SDK 开发与运行指南：从 JVM 工作流 Bundle 到发布全流程
Apache Airflow Java SDK 开发与运行指南从 JVM 工作流 Bundle 到发布全流程【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflowApache Airflow Java SDK 是 Airflow 为 JVM 生态提供的语言 SDK允许开发者用 Java 或其他 JVM 兼容语言编写工作流任务并以可被 Airflow 消费的 Bundle 形式交付执行。本文以java-sdk/README.md为骨架结合仓库中的源码、示例与配置系统讲解 SDK 的模块构成、构建与依赖校验、示例运行方式、任务编写 API、执行期通信协议、兼容性矩阵以及从本地发布到 Maven Central 的完整发布与投票流程。读完本文你将掌握如何构建 SDK、用 Java 编写并打包工作流、将其接入 Airflow 的 Java 队列以及如何独立核验一个 Java SDK 发布候选。一、SDK 定位与运行环境Java SDK 是一个面向JVM的 Apache Airflow 开发套件。你可以使用任意 JVM 兼容语言编写工作流 Bundle再由 Airflow 消费执行结果。SDK 本身与执行期逻辑使用Kotlin实现但公开 API 面向 Java仓库中捆绑了一个使用 Java 调用 SDK 的端到端示例并在scala_spark_example/中提供了 Scala Spark 的使用示例见 ScalaSparkExample.scala。运行环境要求SDK 运行时要求Java 11 或更高版本可选组件与开发工具可能有更高要求具体见 airflow-jvm-conventions.gradle.kts 中定义的 toolchain服务端侧要求Airflow 3.3 及以上版本且 supervisor 通信 schema 版本为2026-06-16见 capabilities.yaml 与gradle.properties中的airflowSupervisorSchemaVersion。更多 Java SDK 的使用细节可参见 Airflow 官方文档Authoring and Scheduling下的 Java SDK 章节。二、仓库布局一个 JVM 多模块 Gradle 工程java-sdk/是一个多模块 Gradle 工程各模块职责清晰源码结构见 java-sdk/settings.gradle.kts模块职责sdk/核心库公开 APIorg.apache.airflow.sdk与内部执行层org.apache.airflow.sdk.execution含Server、Client、Bundle、DagDef、Builder等核心类型processor/注解处理器为被Builder.Dag标注的类生成*Builder类BuilderProcessor.kt基于 kaptplugin/Gradle 插件org.apache.airflow.sdk提供bundle打包任务、Manifest 属性注入与verifyBundleMainClass校验bom/Bill of Materials POM便于消费者以统一版本导入全部 SDK 构件slf4j/SLF4J 日志 Provider将 SLF4J 调用路由到 Airflow 日志存储jul/java.util.loggingHandler将 JUL 记录路由到 Airflow 日志存储对应AirflowJulHandler.ktjpl/Java Platform Logging ProviderSystem.LoggerJEP 264路由 JPL 调用对应AirflowSystemLoggerFinder.ktlog4j2/Log4j 2 Appender路由 Log4j 2 事件到 Airflow 日志存储对应AirflowLog4jAppender.javaexample/端到端示例 Bundle注解 API 接口 APIJava 源码buildSrc/共享 Gradle 约定插件Java 版本、lint、格式化等Python 侧的 JVM 启动协调器不在本目录内位于task-sdk/src/airflow/sdk/coordinators/java/JavaCoordinatorSubprocessCoordinator子类task-sdk/tests/coordinators/java/Python 侧单元与集成测试。三、构建 SDK 与依赖校验构建整个 SDK 只需在java-sdk/目录下执行./gradlew build依赖变更审查gradle/verification-metadata.xml为依赖、插件及其元数据固定了 SHA-256 校验和。只要该文件存在Gradle 会自动启用依赖验证并默认采用strict模式因此普通的./gradlew build就已经在执行校验。strict模式会在以下两种情况下失败校验和不匹配某个构件在元数据中完全没有条目——这通常是升级依赖时最常见的情况意味着构建解析到了元数据未描述的内容。仓库层面还有两条约定仓库集中在settings.gradle.kts中声明且**动态版本dynamic与变化版本changing**对项目依赖配置是被禁止的插件 marker 与 detached 配置除外需要手工固定版本buildSrc/自行声明仓库但其依赖同样被该元数据覆盖。更新依赖或插件时应从可信网络重新生成元数据。注意任务列表必须覆盖 CI 运行的全部内容因为只有被调用任务实际解析到的依赖才会被记录./gradlew --write-verification-metadata sha256 --refresh-dependencies \ build \ :sdk:dokkaGeneratePublicationHtml :sdk:dokkaGeneratePublicationJavadoc \ sourceTarball checksumSourceTarball \ publishToMavenLocal -PskipSigningtrue几点说明不带-PskipSigningtrue时签名任务会失败但 Gradle 仍会从部分运行中写出元数据重新生成只会追加版本升级后过期的条目需要手工删除生成文件只记录仓库当时提供的内容并不代表这些字节可信务必把新坐标与校验和与依赖官方发布信息交叉核对生成元数据不覆盖example/、scala_spark_example/、kubernetes-tests/lang_sdk/java_example/的构建、foojay resolver 自动供应的 JDK以及:sdk:syncSupervisorSchema拉取的 Supervisor Schema。构建文档./gradlew dokkaGenerate该命令使用 Dokka 构建 Java SDK 文档同时生成 HTML 表示与 Javadoc含:sdk:dokkaGeneratePublicationHtml与:sdk:dokkaGeneratePublicationJavadoc两个变体。四、端到端运行示例要让示例真正跑起来需要完成 SDK 构件发布、Bundle 打包、Airflow 队列配置与连接/变量准备四个环节。1. 发布 SDK 到本地 Maven 仓库./gradlew publishToMavenLocal -PskipSigningtrue构建成功后~/.m2/repository/org/apache/airflow/下会出现airflow-sdk、airflow-sdk-processor、airflow-sdk-bom、airflow-sdk-gradle-plugin等目录。2. 打包示例 Bundle进入示例工程并执行 bundle 任务# 已进入 example 目录gradlew 在父目录 cd example ../gradlew bundleBundle 会输出到example/build/bundle。打包由 Gradle 插件AirflowSdkPlugin.kt负责包括 bundle 任务、Manifest 属性注入与verifyBundleMainClass校验。3. 放置 DAG 文件将带 stub 任务的 DAG 放到 Airflow 能找到的位置仓库提供了现成示例java_examples.py。4. 配置 Airflow 的 Java 队列确保 Airflow 任务 worker 所在环境能执行java命令然后配置[sdk]段的协调器把java队列的任务路由给 Java 执行export AIRFLOW__SDK__COORDINATORS{ java: { classpath: airflow.sdk.coordinators.java.JavaCoordinator, kwargs: {jars_root: [/opt/airflow/java-sdk/example/build/bundle]} } } export AIRFLOW__SDK__QUEUE_TO_COORDINATOR{java: java}classpath指向 Python 侧JavaCoordinator的完整导入路径kwargs.jars_root是扫描 JAR Bundle 的目录列表QUEUE_TO_COORDINATOR建立队列名 → 协调器名的映射将java队列路由到名为java的协调器条目。5. 准备连接与变量示例 DAG 依赖的 Connection 与 Variable 可通过环境变量注入export AIRFLOW_CONN_TEST_HTTP{ conn_type: http, login: user, password: pass, host: example.com, port: 1234, extra: {param1: val1, param2: val2} } export AIRFLOW_VAR_MY_VARIABLE123五、编写任务注解 API 与接口 APISDK 提供两套编写任务的 API注解驱动的声明式 API以及面向底层编排的接口 API。1. 注解 API推荐org.apache.airflow.sdk.Builder容器类Builder.kt定义了三枚注解由processor模块的BuilderProcessor.kt在编译期生成*Builder类注解目标参数说明Builder.Dag类id默认取类名、to生成的 Builder 类名默认类名 Builder标注一个 DAG 定义类处理器为其生成FooBuilder.build()返回装配好的DagDefBuilder.Task方法id默认取方法名将方法标注为任务定义Builder.XCom方法参数task默认取参数名将参数标记为来自指定任务的 XCom 输入仓库中的 AnnotationExample.java 展示了完整用法Builder.Dag(id java_annotation_example) public class AnnotationExample { Builder.Task(id extract) public long extractValue(Client client) throws InterruptedException { var pythonXcom client.getXCom(python_task_1); // 读取 Python 任务推送的 XCom var connection client.getConnection(test_http); // ... 业务逻辑 return new Date().getTime(); // 返回值自动作为 return_value XCom 推送 } Builder.Task(id transform) public long transformValue(Client client, Builder.XCom(task extract) long extracted) { var variable client.getVariable(my_variable); return new Date().getTime(); } }要点任务方法参数中的Client由运行时自动注入用Builder.XCom标注的参数会在任务执行前自动拉取上游任务同名或task指定的 XCom方法的返回值会自动以return_value为键推送为 XCom供下游任务消费常量XCOM_RETURN_KEY定义在 Client.kt若某参数声明为原始类型如long而对应 XCom 从未被推送会抛出MissingXComException——改用装箱类型如Long即可接收null见 Client.kt 的MissingXComException构造逻辑。示例中还演示了重试感知load任务在第一次执行context.ti.tryNumber 1时故意抛异常由于 Java SDK 会在ti_context.should_retry置位时返回RetryTask而非终态的FAILEDsupervisor 会把任务标记为UP_FOR_RETRY重试后任务正常完成。同时concurrent任务验证了单个 supervisor 通道可以承载跨线程并发客户端调用8 线程 × 32 次getConnection。2. 接口 API底层编排当需要底层控制时可直接使用 DagDef.kt 中的类型DagDefDAG 定义ID 只能包含 ASCII 字母数字、横线、点或下划线且在一个 Bundle 内唯一通过addTask(id, Class)链式注册任务TaskDef单个任务定义ID 实现Task接口的类类需有公共无参构造器Task接口实现execute(context, client)方法抛出的任何异常都会将任务实例标记为失败。var dag new DagDef(java_etl) .addTask(extract, Extract.class) .addTask(load, Load.class);3. Bundle 与进程入口Bundle.kt 定义Bundle为当前 JVM 进程可执行的全部DagDef的不可变快照DAG ID 重复会抛出IllegalArgumentExceptionBundleBuilder接口负责收集 DAG 并提供build()。进程入口通过Server启动见 ExampleBundleBuilder.javapublic class ExampleBundleBuilder implements BundleBuilder { Override public IterableDagDef getDags() { return List.of( InterfaceExampleBuilder.build(), AnnotationExampleBuilder.build(), XComCastingExampleBuilder.build()); } public static void main(String[] args) { var bundle new ExampleBundleBuilder().build(); Server.create(args).serve(bundle); } }ServerServer.kt是 SDK 的运行核心Server.create(args)解析两个由 Airflow 自动注入的命令行参数--comm host:port任务执行消息通道与--logs host:port日志转发通道无需手工构造serve(bundle)是阻塞入口对serveAsync的包装连接协调器后分发任务执行请求当协调器关闭连接通常在一次任务实例执行后时进程退出进程启动时会同时打开--comm与--logs两个 socket随后等待首帧消息若收到StartupDetails则按dag_idtask_id查表调用用户任务方法若收到ErrorResponse则抛出ApiError。4. 运行时 Client API任务方法中注入的ClientClient.kt封装了与 Airflow supervisor 的通信所有读写默认自动限定在当前 DAG run 与任务实例范围内方法作用getConnection(id)读取 Airflow 连接存储中的连接返回Connection含 id/type/host/schema/login/password/port/extra 字段连接不存在或调用失败抛ApiErrorgetVariable(key)读取 Airflow 变量未设置时返回nullgetXCom(key, dagId, taskId, runId, mapIndex, includePriorDates)读取其他任务推送的 XCommap_index为null时对 mapped 任务返回按 map index 升序聚合的集体结果列表setXCom(key, value)推送 XCom 供下游读取值必须可 JSON 序列化六、执行架构与线协议JavaCoordinator与 JVM 子进程的协作流程如下详见 README Contributing 一节与 coordinator.py启动 JVMAirflow supervisor 判定任务应在 Java 队列运行后调用JavaCoordinator.execute_task()Python。该协调器扫描jars_root构建 classpath并执行java -cp jars MainClass --commhost:port --logshost:port连接建立Server.kt启动后立即连接两个 socket下发任务supervisor 发送StartupDetailsMessagePack 消息JVM 读取后按dag_idtask_id找到匹配任务并调用用户任务方法运行期请求执行期间 JVM 通过 comm 通道向 supervisor 发送请求GetVariable、GetConnection、GetXCom、SetXCom 等并等待响应结束任务完成或抛异常时JVM 发送TaskState消息并关闭 socket进程随之退出。关键实现细节Comm.kt帧格式所有帧均为4 字节大端长度前缀 MessagePack 载荷请求/响应关联每个请求获得递增的id通过pending表ConcurrentHashMapInt, CompletableDeferredIncomingFrame与读循环协程配对响应communicateT()对ErrorResponse自动抛ApiError帧大小防护入站帧上限取协议规定的Frame.MAX_FRAME_LENGTH与堆内存 1/8MAX_HEAP_FRACTION_PER_FRAME 8随-Xmx自适应的较小值超限帧在分配内存前即被拒绝将不可恢复的 OOM 转化为可捕获的FrameProcessingException流式解码ChannelFrameInput以 64 KiB 分块喂给 MessageUnpacker避免超大帧一次性大分配长度前缀承诺的字节数多于实际载荷under-run会被视为流失步日志转发SDK 自身而非用户代码产生的日志消息通过--logssocket 转发由 supervisor 追加到 Airflow 日志存储。对应实现见execution/Logger.kt与各日志桥接模块SLF4J/JUL/JPL/Log4j 2。线协议由 task-sdk/src/airflow/sdk/execution_time/schema/schema.json 统一定义新增消息类型需要同时修改 Python 侧schema.json与 JVM 侧execution/Comm.ktexecution/Client.kt。关于协调器架构与各语言 SDK 共享的核心集成面可进一步阅读 airflow-core/adr/lang-sdk 下的架构决策记录ADR。从 Python 协调器源码看JavaCoordinator支持以下配置项见 coordinator.py 类 docstring 与字段定义参数默认值说明java_executablejavajava命令路径默认依赖$PATHjvm_args[]额外 JVM 参数如[-Xmx1024m]jars_root必填至少 1 个扫描 JAR Bundle 的目录列表main_class显式入口类未指定时扫描jars_root寻找带Main-Class元数据的可执行 JAR若存在多个可能不确定task_startup_timeout10 秒等待任务进程启动的最长时间JavaCoordinator._build_execute_task_command()会从 JAR 的META-INF/MANIFEST.MF中读取Main-Class与Airflow-Supervisor-Schema-Version两个条目后者用于核对线协议 schema 版本——Java SDK 打包的 JAR 会自动写入该元数据只要依赖 JAR 原样部署就无需额外处理若重新打包依赖则必须在一个 JAR 中复现该条目。七、兼容性矩阵README 中的兼容性矩阵由 capabilities.yaml 自动生成update-java-sdk-readme-matrixpre-commit 钩子负责重生成请勿手工编辑表格符合 contributing-docs/30_new_language_sdk.rst 定义的 Language SDK 一致性规范。当前 SDK 的能力概览对应 Airflow 3.3、supervisor schema2026-06-16TaskInstance 状态支持MUST3.3success、failed、up_for_retry通过RetryTask、removed未支持skipped、deferred、up_for_reschedule、awaiting_input运行时尚未发出对应消息。运行时能力支持MUST3.3mixed-lang-stub-targettask.stub、task-loggingSLF4J JPL 桥接到任务日志、xcom-read-write、connection-read、self-contained-bundleAirflow 元数据内嵌于 jar 构件未支持variable-read-write目前仅getVariable尚无写通道、retry-policy、task-state-store、asset-state-store、asset-event-emit、asset-event-read。Native-Dag 编写native-dag-authoring尚未实现因此task-args、dag-params、taskflow-dependencies、branching、dag-test、task-group、dynamic-task-mapping、asset-inlets-outlets、asset-scheduling、object-store等原生能力均标注为 n/a仅在 native-dag-authoring 支持后才适用。在 Airflow 侧混合语言场景通过task.stub声明 stub 任务并指定队列由 supervisor 把对应队列的任务路由给 Java 协调器执行Java 任务通过 XCom 与 Python 任务交换数据示例 DAG java_examples.py 中可见python_task_1/python_task_2与 Java 任务的 XCom 互通。八、发布流程SDK 通过 ASF Nexus staging 仓库发布到 Maven Central。凡发布到 Maven Central而非 Snapshots的版本均视为正式发布含 alpha、beta 等每次发布都需 PMC 投票仅-SNAPSHOT构建可免投票发布。前置条件具备访问 repository.apache.org 的 ASF committer 账号以及已加入项目 KEYS 文件并上传到公共密钥服务器的 GPG 密钥。1. 版本号编辑gradle.properties设置projectVersionVERSION或对单次命令用-PprojectVersionVERSION覆盖。版本字符串需符合 Maven 版本顺序规范例如 beta 1 写作1.0.0-beta1。main分支在发布之间应保持在-SNAPSHOT版本snapshot 排序在VERSION之后、GA 之前因此 beta 之后无需额外 bump。2. 打 RC 标签git tag -s java-sdk/VERSION-rcN -m Java SDK VERSION RC N git push upstream java-sdk/VERSION-rcNRC 编号保留在标签名中投票失败只需递增到下一 RC构件版本本身不带 RC 后缀。投票前需先推送标签便于评审者检出确切的待投票源码。3. 本地核验 POMrm -rf ~/.m2/repository/org/apache/airflow/ # 清空旧构件 ./gradlew publishToMavenLocal -PskipSigningtrue less ~/.m2/repository/org/apache/airflow/airflow-sdk/*/airflow-sdk-*.pom less ~/.m2/repository/org/apache/airflow/airflow-sdk-bom/*/*.pom less ~/.m2/repository/org/apache/airflow/airflow-sdk-processor/*/airflow-sdk-*.pom less ~/.m2/repository/org/apache/airflow/airflow-sdk-gradle-plugin/*/airflow-sdk-*.pom less ~/.m2/repository/org/apache/airflow/sdk/org.apache.airflow.sdk.gradle.plugin/*/*.pom核对各 POM 的坐标、描述、license、SCM 与 organization 字段。4. 对本地仓库做发布演练rm -rf /tmp/local-maven-repo ./gradlew publish -PmavenUrlfile:///tmp/local-maven-repo -PskipSigningtrue ls /tmp/local-maven-repo/org/apache/airflow/输出应与上一步~/.m2中的构件一致。本地演练不需要签名如需测试签名按下一节配置 GPG 私钥与口令并去掉-PskipSigningtrue。5. 发布到 ASF Nexus staging将凭据写入~/.gradle/gradle.properties避免进入 shell 历史mavenUsernameyour-asf-nexus-token-username mavenPasswordyour-asf-nexus-token-password signing.passwordyour-gpg-key-passphrase然后暂存并关闭发布根build.gradle.kts应用的 Gradle Nexus Publish Plugin 会把所有模块聚合进同一个staging 仓库无需逐模块协调./gradlew publishToApache closeApacheStagingRepository \ --no-configuration-cache \ -Psigning.key$(gpg --armor --export-secret-keys your-gpg-key-fingerprint)注意三点签名密钥通过命令行传入因其含换行符不适合放在 Gradle properties 文件也可用环境变量提供凭据ASF_NEXUS_USERNAME、ASF_NEXUS_PASSWORD、SIGNING_KEY、SIGNING_PASSWORD适合 CI项目全局启用 configuration cache但 staging 任务publishToApache、closeApacheStagingRepository、releaseApacheStagingRepository通过 Nexus REST API 工作且不兼容 configuration cache因此发布命令必须加--no-configuration-cache。发布后在 Nexus 的Staging Repositories中打开已关闭的仓库核验所有模块均包含 jar、-sources.jar、-javadoc.jar适用处、.pom与.asc签名并检查Updated by、Uploaded Date、Last Modified字段。6. 上传源码包投票正式针对的是已签名的源码包Maven 构件是 convenience binaries。sourceRelease任务基于已提交的java-sdk源码含LICENSE与NOTICE一次生成签名与校验和./gradlew sourceRelease -PgitRefjava-sdk/VERSION-rcN输出到build/distributions/的三个文件apache-airflow-java-sdk-VERSION-src.tar.gz apache-airflow-java-sdk-VERSION-src.tar.gz.asc apache-airflow-java-sdk-VERSION-src.tar.gz.sha512注意源码包刻意省略 Gradle wrapper 脚本gradlew、gradlew.bat与gradle/wrapper/gradle-wrapper.jar——ASF 源码发布不得包含编译产物见 LEGAL-570而缺少 jar 的脚本没有用处gradle/wrapper/gradle-wrapper.properties会保留以固定 Gradle 版本与发行版校验和供重新生成 wrapper 时核验。将三个文件复制到 ASF distdev仓库并提交svn checkout https://dist.apache.org/repos/dist/dev/airflow dist-dev-checkout cd dist-dev-checkout mkdir -p java-sdk/VERSION-rcN cp path-to/java-sdk/build/distributions/apache-airflow-java-sdk-VERSION-src.tar.gz* \ java-sdk/VERSION-rcN/ svn add --parents java-sdk/VERSION-rcN svn commit -m Add Apache Airflow Java SDK VERSION-rcN source release candidate7. 发起投票向devairflow.apache.org发送[VOTE]邮件链接 git 标签与提交、dist/dev中的源码包、已关闭的 Nexus staging 仓库及KEYS文件。投票至少开放 72 小时。README 提供了完整模板subject 为[VOTE] Release Apache Airflow Java SDK VERSION based on VERSION-rcN正文列出 Maven 构件清单、Git 信息、源码包与 convenience binaries URL、投票选项。发送前检查所有构件能通过 BOM 交叉解析、Changes since rc 与截止时间已填写、对渲染后的邮件执行grep email.txt确认无输出任何匹配都说明占位符未填完。8. 独立核验发布候选任何社区成员都应在投票前独立核验候选核验清单如下校验和sha512sum -c apache-airflow-java-sdk-VERSION-src.tar.gz.sha512签名下载并导入KEYS文件用gpg --verify验证.asc签名与 git 标签比对解包 tarball与标签的干净检出做diff -rq除.gitattributesexport-ignore排除的文件gradlew、gradlew.bat、gradle-wrapper.jar、scripts外应无差异解包后的顶层目录应为apache-airflow-java-sdk-version不带仅出现在压缩包文件名中的-src后缀无二进制文件ASF 源码发布不得含编译产物用file扫描非文本文件应无输出从源码构建用本地 Gradle 依gradle-wrapper.properties中的distributionUrl与distributionSha256Sum重新生成 wrapper再./gradlew buildstaged 二进制冒烟测试从一个临时工程指向 staging 仓库 URL声明org.apache.airflow:airflow-sdk-bom:VERSION依赖确认传递构件含airflow-sdk-jpl可解析、示例 Bundle 能构建对应脚本见 scripts/ci/smoke-test-staged-binaries.sh 与 scripts/ci/verify-source-release.sh。9. 投票成功后的收尾回复[RESULT][VOTE]统计然后releasestaging 仓库同步到 Maven Central 需数小时不重新构建或签名./gradlew releaseApacheStagingRepository --no-configuration-cache将源码包从dist/dev移至dist/releasesvn mv在同一被投票的提交上打最终版本标签java-sdk/VERSION保留 RC 标签以便追溯发布 GitHub release附带被投票且已签名的源码构件gh release create预发布版本加--prerelease核验标签用--verify-tag等待约 1 小时Maven Central 同步后发送纯文本[ANNOUNCE]邮件到usersairflow.apache.org抄送dev并在 ASF Committee Report Helper 中记录发布触发Publish Docs to S3workflow 发布 API 文档确认https://airflow.apache.org/docs/java-sdk/stable/可解析且/docs/java-sdk/重定向到它。若投票失败关闭投票、在 Nexus 中dropstaging 仓库、删除dist/dev候选、修复问题后切下一 RC...-rc2。发布版本号不变仅标签中的 RC 计数递增。九、测试与编码规范运行测试# 运行全部 JVM 测试 ./gradlew test # 运行指定测试类 ./gradlew :sdk:test --tests org.apache.airflow.sdk.execution.CommTestPython 协调器测试必须通过 Breeze 运行不要在宿主机直接跑 pytestbreeze testing task-sdk-tests -- task_sdk/coordinators/java端到端测试需要真实 Airflow 环境E2E_TEST_MODEjava_sdk uv run --project airflow-e2e-tests pytest \ tests/airflow_e2e_tests/java_sdk_tests/ -xvs编码规范所有 SDK 与 processor 源码均为KotlinJava 是公开 API 目标而非实现语言保持sdk/src/main/kotlin/公开 API 面不混入内部实现细节内部实现属于execution/子包注解处理器BuilderProcessor.kt使用 kapt新增注解时在Builder.kt定义、在BuilderProcessor.kt处理并在processor/src/test/kotlin/添加 golden-output 测试Python 协调器继承SubprocessCoordinator除_build_execute_task_command提供的接口外不要从 Python 深入 JVM 进程内部提交前运行./gradlew ktLintCheck spotlessCheck或ktLintFormat spotlessApply项目强制 Kotlin 与 Java 格式所有新文件必须带 Apache License 头。常见开发任务新增一个Client方法如新的 Airflow API 调用若消息类型是新的从schema.json重新生成 POJO在execution/Comm.kt或新文件添加 Kotlin 请求/响应数据类在公开Client.kt添加委托给execution/Client.kt做 wire 调用的方法在sdk/src/test/kotlin/.../ClientTest.kt编写 mock socket 层的单元测试若变更用户可见同步更新airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst。新增注解在Builder.kt定义注解接口 → 在BuilderProcessor.kt生成对应代码 → 在BuilderTest.kt添加期望生成输出的测试 → 更新java.rst的注解表。修复组帧或协议 bug聚焦execution/Comm.kt与execution/Frame.ktCommTest.kt覆盖编解码往返修复前先添加复现该 bug 的回归测试。PR 检查清单运行./gradlew build testJVM与对应 pytest 套件Python 协调器确认示例 Bundle 仍可编译按运行示例一节做到打包步骤若schema.json变更确认 JVM 与 Python 两侧都能处理新/改字段为每项行为变更添加或更新测试对task-sdk/的用户可见变更在airflow-core/newsfragments/下添加 newsfragment。十、小结Apache Airflow Java SDK 让 JVM 团队以惯用的 Java/Kotlin/Scala 方式编写 Airflow 任务通过Builder.Dag/Builder.Task/Builder.XCom注解声明式描述 DAG用 Gradle 插件一键打成自包含 Bundle由JavaCoordinator以子进程方式拉起 JVM 并通过 4 字节长度前缀 MessagePack 的 comm 通道与 supervisor 协作天然支持与 Python 任务的 XCom 互通和task.stub混合语言调度。其发布流程遵循 ASF 规范从本地 POM 核验、staging 发布、源码包签名、PMC 投票到最终同步 Maven Central 均有清晰的步骤与可执行的验证清单为在 Java 生态中落地 Airflow 提供了完整链路。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考