Support DagBundle based deployment for Coordinator by jason810496 · Pull Request #70805 · apache/airflow

jason810496 · 2026-07-31T02:27:55Z

closes: Java-SDK artifact deployment strategies #66334

Why

Language-SDK subprocess coordinators (Java, Executable, Node) required a hand-managed filesystem root (jars_root / executables_root / bundles_root) for compiled artifacts. Users should be able to deploy those artifacts by existing DagBundle mechanism as well.

How

Add an optional dag_bundle_name kwarg on the SubprocessCoordinator base, giving three ways to locate artifacts:
- explicit root set: scan that path (unchanged legacy behavior).
- dag_bundle_name set: scan that configured bundle's path.
- neither set: scan the task's own bundle, co-located with the Python stub Dag.
Resolve roots in the base and expose them through _get_scan_roots() while keeping the existing _build_execute_task_command(what) hook signature, so a subclass only declares its explicit-root kwarg through _explicit_artifact_roots.
Materialize the bundle through airflow.sdk.execution_time.bundles.initialize_ti_bundle, a new module split out of task_runner so starting a language-SDK task does not import the task runner into the supervisor.
Validate at construction that an explicit root and dag_bundle_name are not both set, and that a named bundle is configured, via a new DagBundlesManager.is_bundle_configured that reads configured names without importing any bundle class.

Notes

Version semantics: a co-located bundle is pinned to the run's bundle_info.version because it is the same artifact the run was created from; a separately named artifact bundle resolves to the version current when the task starts. Either way the bundle is materialized at a concrete version, so BundleVersionLock protects the same tree the subprocess scans.

Was generative AI tooling used to co-author this PR?

Yes, with help of Claude Code (Opus 4.8) following the guidelines

uranusjr · 2026-08-07T00:23:38Z

Keep the subclass _build_execute_task_command signature unchanged: execute_task stashes the task's bundle_info for the duration of the call, and _scan_roots resolves the scan root on demand.

But the signature still changed, a new root has been added. Either this also needs to go away or made compatible, or we should just bite the bullet and pass more arguments instead.

jason810496

Thanks for the review.

Language-SDK coordinators previously required a hand-managed filesystem root (jars_root/executables_root/bundles_root) for compiled artifacts, kept outside Airflow's Dag-bundle machinery and prone to drifting from the Dags they serve. Allowing artifacts to be located through a configured Dag bundle -- or the task's own bundle when none is named -- lets them be delivered and versioned the same way as everything else.

Language-SDK subprocess coordinators materialized artifact bundles without the version-lock protection the Python task path holds, so bundle cleanup could delete a version a running JVM was still reading from. Resolution now holds the version lock for the subprocess lifetime, mirroring the Python task path, and materialization failures surface in the task log rather than the worker log. Classification moves to the base coordinator so a subclass that adds no explicit-root kwarg falls back to the task-bundle default instead of failing every task, the Java coordinator now requires an explicit entrypoint when it has to scan a whole bundle, and configured-bundle lookup checks names only so an unrelated unimportable bundle no longer breaks coordinator construction.

When a subprocess coordinator locates artifacts through a Dag bundle rather than an explicit root, JavaCoordinator previously rejected a missing main_class outright. That made the co-located default unusable without extra configuration and left the entrypoint no more determinate than an auto-detected scan would be. Auto-detect over the resolved bundle now handles that case, with the remaining cross-bundle ambiguity deferred to an IMPORT_ERROR-stage check under AIP-85. A named Dag bundle resolves to its latest version, so its version was unset and the version lock held around the subprocess silently did nothing, letting bundle cleanup delete artifacts a running task was still reading. The concrete current version is now pinned before locking.

A language-SDK task start needs a bundle materialized on disk before the subprocess can be launched, but reaching that through the task runner made every such start pay for the operator tree, pydantic, and OTel plumbing that the supervisor side never uses. The bundle lifecycle now lives on its own, so the coordinator depends on the Dag-bundle machinery alone. The bundle-name check a coordinator runs at construction and the class-importing parse the manager runs at startup previously read the configuration independently, so the two could disagree about what is configured. Keying the parsed configuration on the configuration itself lets both share one snapshot while a configuration change is still picked up. Subclasses had to declare the name of their explicit-root kwarg separately from its value, which is two things to keep in step for one piece of information.

Splitting the read across a keyed builder and a wrapper meant two names for one idea, and the empty case leaked out as a module-level constant that existed only to be returned. One function now answers "what is configured", parsed once per process so both readers of it see the same answer. Parsing once per process means a later configuration change is not observed. That only arises in tests, which switch configuration freely, so overriding configuration there drops the memo rather than silently serving a stale one.

External language SDK coordinators can implement the documented command hook. Keeping that contract stable avoids unexpected-keyword failures when upgrading while allowing Dag-bundle-backed artifact discovery.

Module-level imports keep dependency analysis reliable and follow Airflow's import-placement contract without changing bundle access behavior.

Bundle resolution does not need mutable instance state, while overwriting scan roots can make a command build against another task's artifacts.

An explicitly empty root is likely a configuration or templating error. Treating it as omitted can silently scan the task's Dag bundle and run unintended artifacts.

An empty string normalizes to the current directory, so it bypassed the empty-roots check and scanned the worker's working directory.

jason810496

Thanks for the review.

uranusjr

One nit, otherwise lgtm.

jason810496 requested review from amoghrajesh, ashb, ephraimbuddy, jedcunningham, kaxil and uranusjr as code owners July 31, 2026 02:27

boring-cyborg Bot added area:coordinator Coordinator: The interface to spawn Lang-SDK subprocesses area:dag-processor area:task-sdk labels Jul 31, 2026

jason810496 self-assigned this Jul 31, 2026

kaxil reviewed Jul 31, 2026

View reviewed changes

jason810496 marked this pull request as draft July 31, 2026 15:24

jason810496 force-pushed the feature/coordinator/dag-bundle-name branch from 1b76bfe to 182919a Compare August 1, 2026 15:29

jason810496 marked this pull request as ready for review August 3, 2026 08:54

jason810496 mentioned this pull request Aug 4, 2026

Report language-SDK tasks running while the runtime starts up #71075

Closed

1 task done

jason810496 added this to the Airflow 3.4.0 milestone Aug 5, 2026

jason810496 mentioned this pull request Aug 5, 2026

Coordinator: detect ambiguous entrypoints and duplicate dag_ids at import time #71134

Open

jason810496 force-pushed the feature/coordinator/dag-bundle-name branch 2 times, most recently from bc97b03 to 12de628 Compare August 5, 2026 06:34

jason810496 mentioned this pull request Aug 5, 2026

Extract supervisor heartbeat logic into a standalone Heartbeater #71102

Draft

1 task done

jason810496 force-pushed the feature/coordinator/dag-bundle-name branch from 12de628 to c7ab92a Compare August 6, 2026 03:19

uranusjr reviewed Aug 6, 2026

View reviewed changes

Comment thread airflow-core/src/airflow/dag_processing/bundles/manager.py

uranusjr reviewed Aug 6, 2026

View reviewed changes

Comment thread task-sdk/src/airflow/sdk/execution_time/task_runner.py Outdated