Move the Dag file processor's shared plumbing into a base class by jason810496 · Pull Request #73972 · apache/airflow

jason810496 · 2026-09-30T17:59:41Z

Stack (bottom to top): #73970, #73971, #73972, #73973, #73974, #73975, #73976, #73977, #74030, #74031, #74032, #74033, #74067

Why

A Lang-SDK artifact is parsed by processes other than the Python parse child: the native Dag parse (#73842) and, later in this stack, the task-handler probe (ADR-0012, ADR-0013). They share the parse log forwarding, the request handlers, readiness and cleanup with DagFileProcessorProcess, so that plumbing moves into a base class they can all subclass.

What changes

No behavior change.
BaseDagFileProcessorProcess holds the parse log forwarding (task. -> dag_processor.), the request handlers, readiness and cleanup. Its logger_filehandle is optional, and close() skips it when unset.
DagFileProcessorProcess keeps only how the Python parse child starts and what it is sent first, and it still requires logger_filehandle.
Parse native Lang-SDK Dags with their coordinator's runtime #73842 will pick up this same change and rebase onto it unchanged.

How to test

uv run --project airflow-core --with-editable shared/secrets_masker pytest \
  airflow-core/tests/unit/dag_processing/test_processor.py \
  airflow-core/tests/unit/dag_processing/test_manager.py -q

Ran: test_processor.py 124 passed; test_manager.py 175 passed, 5 skipped (Postgres-only).

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

Yes (please specify the tool below)

Generated-by: Claude Code (Opus 5.5) following the guidelines

ADR-0013 locates Lang-SDK task handlers through a named Dag bundle, so the unversioned jars_root, executables_root and bundles_root directories go away instead of living beside it. dag_bundle_name becomes task_handler_bundle_name because the native-Dag work will use the old name with a different meaning. It has not shipped in a release, so there is no shim. A coordinator reads handlers from the named bundle at its current version, or from the task's own Dag bundle at the run's version, pinned for the task either way. CoordinatorManager.from_config validates the name of every spec without constructing coordinators, so a typo fails at config load instead of on the first task.

The compose e2e and the k8s lang-SDK test configured coordinators with the explicit artifact roots, which no longer exist. The Java e2e needs three bundles because one bundle is one classpath, and its three coordinators each need their own dependencies. Artifacts stay mounted on the workers only, because only the workers run them; the other components need just the bundle registration for the name to resolve. The k8s init container keeps staging from S3 because the S3 download drops the Go binary's execute bit, which the coordinator requires.

The Go, Java and TypeScript guides, the bundle specs, the READMEs and the contributor guide still pointed users at the removed jars_root, executables_root and bundles_root kwargs. A Dag bundle now decides what a coordinator sees, so two of its limits reach users. For Java, every JAR in the bundle goes on one classpath, so conflicting dependency versions cannot share a bundle, and a second Main-Class in it makes the entry point ambiguous. For Go, the coordinator skips files without the execute bit, so an object-store Dag bundle that drops it leaves the coordinator with no executable to find.

Review flagged a Java test fixture still named after the removed jars_root kwarg, an e2e helper without a verb, and two execute_task patches that were not autospecced like the ones beside them.

The guides said only the workers need the artifact Dag bundle registered, which reads as permission to register it there alone. The [sdk] configuration is also loaded by the scheduler under KubernetesExecutor, and it is rejected wherever the name is missing, so the bundle belongs in the shared bundle list.

Two sentences rewritten for artifact Dag bundles kept their em-dashes, which the docs style for this work avoids.

In Airflow 3.4 the Dag processor checks a Python Dag's stub tasks against the task handlers the language SDK artifacts register, so a deployment that follows the old worker-only advice fails that check. task_handler_bundle_name only serves those stub tasks, which the docs did not say.

ADR-0013 records at parse time which Language SDK artifact runs each stub task, so task execution can stop scanning artifact bundles. This adds the two tables and their migration. Later layers write and read them. The unique key and the owning-file index use an md5 hash of each path instead of the path. A 2000-character path is over MySQL's 3072-byte key limit and Postgres's 2704-byte btree entry limit. This follows DagPriorityParsingRequest.

The tables as built need an md5 hash column beside each long path, a wider cache_digest, and the repo's constraint names. Later layers read this SQL as the spec, so it now matches the migration.

An ORM update of an artifact's relative_fileloc kept the old hash, so the unique key no longer matched the path. The handler table already resyncs its hash this way.

The handler-to-artifact foreign key has no ON DELETE on purpose, so a cleanup that races with a new binding fails instead of dropping it. Only the cascade from dag was tested.

The old docstring read as if the row were the artifact. A row is the binding of one stub task to an artifact.

The id and create date of migration 0142 were not made by alembic. They are now alembic's own rev_id() and a real timestamp, and the hooks updated the revision map and migration reference to match.

A task handler declaration now says whether it binds stub arguments positionally, by name, or by name or as a whole. The fast path checks a changed Python file against the stored declaration without running the runtime, so it needs the mode as well as the params.

The migration now stores the declaration's binding mode, and later layers read this SQL as the spec.

airflow-core takes session keyword-only so that it can never be passed in the wrong position, and review flagged the new test helpers for taking it positionally.

The fast path trusts an artifact's fingerprint as the cache key, but a binding stored only what its own file's probe asked about, so one file probing a rebuilt or colliding artifact left another file's bindings stale. Once the runtime is asked for every handler, its answer depends on the artifact alone, so it belongs beside the fingerprint and is written by the same probe. The binding's own copy then has no reader.

The migration now caches the probe answer on the artifact row, and later layers read this SQL as the spec.

Some artifacts carry no digest: a Go bundle packed before digests.cache existed, or a third-party coordinator that computes none. Such a candidate is still probed and bound, so it needs an artifact row. A NULL digest never matches, so the candidate is probed on every parse.

The migration now lets an artifact store no digest, and later layers read this SQL as the spec.

The native Dag parse and the task-handler probe start parse children other than the Python one, and they need the same parse log forwarding, request handling, readiness and cleanup. No behavior change.

The base class makes logger_filehandle optional for parse children other than the Python one, but DagFileProcessorProcess still requires it, so nothing in this layer reached the close path without one.

jason810496 · 2026-10-06T01:41:20Z

Closing this: the change landed on main another way. #74040 moved the Dag file processor's shared plumbing into a base class, and #74043 made the parse client and log handle optional, which was the rest of this PR. The stack above it no longer needs this layer.

boring-cyborg Bot added the area:DAG-processing label Sep 30, 2026

jason810496 added this pull request to stack #73978 September 30, 2026 17:59

This was referenced Sep 30, 2026

Replace coordinator artifact roots with task_handler_bundle_name #73970

Merged

Add Lang-SDK task handler binding tables #73971

Draft

jason810496 changed the title ~~jason/core taskhandler refactor/03 parse process base~~ Move the Dag file processor's shared plumbing into a base class Sep 30, 2026

jason810496 force-pushed the jason/core-taskhandler-refactor/02-task-handler-tables branch from 5da7e9f to 5e4c9c7 Compare October 1, 2026 06:26

jason810496 force-pushed the jason/core-taskhandler-refactor/03-parse-process-base branch from 43bb8f6 to c93fea0 Compare October 1, 2026 06:26

jason810496 force-pushed the jason/core-taskhandler-refactor/02-task-handler-tables branch from 5e4c9c7 to 98e861e Compare October 1, 2026 06:30

jason810496 force-pushed the jason/core-taskhandler-refactor/03-parse-process-base branch 2 times, most recently from 72f869f to e9b1ba3 Compare October 1, 2026 10:55

jason810496 force-pushed the jason/core-taskhandler-refactor/02-task-handler-tables branch from 0cafad2 to 9bb714b Compare October 1, 2026 14:44

jason810496 force-pushed the jason/core-taskhandler-refactor/03-parse-process-base branch from e9b1ba3 to 59a5bc2 Compare October 1, 2026 14:44

jason810496 added 7 commits October 2, 2026 02:07

Tidy names and mocks in the artifact bundle tests

b9cde7f

Review flagged a Java test fixture still named after the removed jars_root kwarg, an e2e helper without a verb, and two execute_task patches that were not autospecced like the ones beside them.

Drop em-dashes from the rewritten doc sentences

8997a35

Two sentences rewritten for artifact Dag bundles kept their em-dashes, which the docs style for this work avoids.

jason810496 force-pushed the jason/core-taskhandler-refactor/03-parse-process-base branch from 59a5bc2 to c6ea332 Compare October 2, 2026 05:48

jason810496 force-pushed the jason/core-taskhandler-refactor/02-task-handler-tables branch from 9bb714b to 1db024e Compare October 2, 2026 05:48

jason810496 added 17 commits October 2, 2026 11:28

Add newsfragment for the removed coordinator artifact roots

9d00ead

Mark ADR-0013 as accepted

68f03d6

Match ADR-0013 task handler table SQL to the migration

6762729

The tables as built need an md5 hash column beside each long path, a wider cache_digest, and the repo's constraint names. Later layers read this SQL as the spec, so it now matches the migration.

Keep the artifact path hash in sync on ORM updates

dc15082

An ORM update of an artifact's relative_fileloc kept the old hash, so the unique key no longer matched the path. The handler table already resyncs its hash this way.

Test that a referenced task handler artifact cannot be deleted

ab8f025

The handler-to-artifact foreign key has no ON DELETE on purpose, so a cleanup that races with a new binding fails instead of dropping it. Only the cascade from dag was tested.

Describe LangSDKTaskHandler as a stub task binding

5da25f7

The old docstring read as if the row were the artifact. A row is the binding of one stub task to an artifact.

Regenerate the task handler migration revision id

396cac2

The id and create date of migration 0142 were not made by alembic. They are now alembic's own rev_id() and a real timestamp, and the hooks updated the revision map and migration reference to match.

Add handler_binding to ADR-0013's task handler table SQL

fce1160

The migration now stores the declaration's binding mode, and later layers read this SQL as the spec.

Make session keyword-only in the task handler model test helpers

d18d0af

airflow-core takes session keyword-only so that it can never be passed in the wrong position, and review flagged the new test helpers for taking it positionally.

Move the handler declarations to the artifact in ADR-0013's SQL

0361303

The migration now caches the probe answer on the artifact row, and later layers read this SQL as the spec.

Allow a NULL cache digest in ADR-0013's artifact table SQL

da3372e

The migration now lets an artifact store no digest, and later layers read this SQL as the spec.

Move the Dag file processor's shared plumbing into a base class

db4f847

The native Dag parse and the task-handler probe start parse children other than the Python one, and they need the same parse log forwarding, request handling, readiness and cleanup. No behavior change.

Test closing a Dag file processor that has no parse log file

7e26f0d

The base class makes logger_filehandle optional for parse children other than the Python one, but DagFileProcessorProcess still requires it, so nothing in this layer reached the close path without one.

jason810496 force-pushed the jason/core-taskhandler-refactor/02-task-handler-tables branch from 1db024e to da3372e Compare October 2, 2026 11:36

jason810496 force-pushed the jason/core-taskhandler-refactor/03-parse-process-base branch from c6ea332 to 7e26f0d Compare October 2, 2026 11:36

jason810496 force-pushed the jason/core-taskhandler-refactor/02-task-handler-tables branch from da3372e to 69ca1b8 Compare October 5, 2026 03:37

jason810496 closed this Oct 6, 2026

jason810496 removed this pull request from stack #73978 October 6, 2026 02:27

Navigation Menu

Repository navigation

Move the Dag file processor's shared plumbing into a base class - #73972

Move the Dag file processor's shared plumbing into a base class#73972
jason810496 wants to merge 25 commits into
jason/core-taskhandler-refactor/02-task-handler-tablesfrom
jason/core-taskhandler-refactor/03-parse-process-base

jason810496 commented Sep 30, 2026 •

edited

Loading

Uh oh!

jason810496 commented Oct 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

jason810496 commented Sep 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why

What changes

How to test

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

Uh oh!

jason810496 commented Oct 6, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

jason810496 commented Sep 30, 2026 •

edited

Loading