Reload Dag bundle configuration without restarting processors

[samraj2k]

Summary

Dag processors currently read the bundle list only at startup. A deployment must restart them to add or remove a bundle. In a large deployment, these restarts are difficult when bundles are added frequently.

This PR adds the capability of hot reload using the existing dag_bundle_config_list. The Dag processor checks for changes at the existing bundle_refresh_check_interval. The configuration source returns the complete list of bundles.

A deployment can supply the list through Airflow's existing _CMD or _SECRET configuration mechanism. For example, a command can read a file or call an internal API:

export AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST_CMD='cat /opt/airflow/bundles.json'

Existing literal environment variables and airflow.cfg values keep their startup behavior. Users do not need to rename their environment variables unless they want to enable reload. The ordinary environment variable takes precedence over _CMD and _SECRET.

Behavior

For example, a deployment can publish these lists without restarting the Dag processor:

[finance]        -> Start processing finance
[finance, risk]  -> Start processing risk; continue finance
[risk]           -> Stop processing finance and mark its Dags stale
[]               -> Remove all bundles

• A missing bundle is marked inactive by an unfiltered processor. The existing stale-Dag scan marks its Dags stale.
• A successful read containing [] removes all bundles. Blank output is treated as a failed read, so a command must report a failure rather than return [].
• Failed commands, unavailable secrets, malformed configuration, and an invalid new classpath keep the last accepted list in that process. An update is accepted as a complete list.
• A fresh process has no previous list to use. Invalid or unavailable configuration prevents startup. A valid [] starts with no bundles.
• Use a new bundle name when changing its definition, such as its classpath or repository. Existing bundle instances are reused, and failed construction of a new bundle is retried.
• Processors using --bundle-name stop parsing removed bundles locally, but never deactivate missing bundles globally. An unfiltered processor is needed for global deactivation.

Multiple Dag processors

The approach allows eventual convergence. Processors may see a change at different times. For example:

1. Processor A still sees [finance].
2. Processor B sees [] and marks finance inactive.
3. Processor A marks finance active again, then stops permanently.

Each unfiltered processor reconciles bundle active state with the database on every check, even when its local list has not changed. Processor B can therefore repair that stale write. The state converges when surviving processors see the same list and continue processing; the change is not immediate across all hosts.

Finish existing Dag runs before removing a bundle if those runs need to complete. Tasks and callbacks can still need its definition.

This PR is based on main and can be reviewed independently of #73209 and #73217. It adds no public source class, new configuration option, database migration, or DagFileParseRequest change. The configuration key is marked sensitive to support _CMD and _SECRET, so configuration views that hide sensitive values will also hide this value.

Testing

• Relevant OSS suites in Breeze with PostgreSQL 16 and Python 3.11: 1,214 passed, 3 skipped, 2 deselected. The two deselected isolation tests also fail on unchanged main in the local environment.
• Live tests with an API-fetching command, a secrets backend, multi-team mode, and two filtered processors: 12 successful Dag runs.
• Verified adding, removing, removing all, restoring, failed reads, and restarting with []. Checked service, parser, and task logs for unexpected errors and repeated removal warnings.
• Ruff, core mypy, fast and manual prek checks passed. The full CI matrix was not run locally.

Related: #61645, #71111, #73209, #73217.

---

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

• Yes — Codex (GPT-6)

Generated-by: Codex (GPT-6) following the guidelines

---

Drafted-by: Codex (GPT-6) (no human review before posting)