Multiple dag-processors with partial bundle configs deactivate each other's bundles

[sortega]

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.2.2 (also present on main/3.3.0; the affected code is unchanged)

What happened and how to reproduce it?

Summary

When multiple dag-processor processes are each configured with a partial bundle configuration — i.e. each processor's [dag_processor] dag_bundle_config_list lists only the bundle(s) it owns — the processors repeatedly deactivate each other's bundles and the deployment never converges. This is a natural way to shard DAG parsing across processors (one bundle per processor), and it deadlocks.

Root cause

DagBundlesManager.sync_bundles_to_db() loads all bundles stored in the DB, pops the ones present in this process's config, and then unconditionally marks every remaining stored bundle inactive:

for name, bundle in stored.items():
    bundle.active = False
    bundle.teams = []
    self.log.warning("DAG bundle %s is no longer found in config and has been disabled", name)
    session.execute(delete(ParseImportError).where(ParseImportError.bundle_name == name))

The method implicitly assumes the calling process sees the complete bundle configuration. When each processor only sees its own bundle, every processor treats all other processors' bundles as "no longer in config" and disables them. Processor A disables B's bundle; B then disables A's; on the next cycle they flip again. The log line

DAG bundle <name> is no longer found in config and has been disabled

is emitted continuously for bundles that are, in fact, actively configured on another processor.

Steps to reproduce

1. Configure two dag-processor processes.
2. Give processor A dag_bundle_config_list = [bundle-a] and processor B dag_bundle_config_list = [bundle-b] (disjoint, per-processor configs).
3. Let both run sync_bundles_to_db() (happens on each parse loop startup).
4. Observe the dag_bundle table: bundle-a.active and bundle-b.active flip between True/False on every cycle, and the "no longer found in config and has been disabled" warning is logged for a bundle that is actively configured elsewhere.

Minimal reproduction of the flip-flop (two managers with disjoint config, no error path involved):

AFTER B syncs:       [('bundle-a', False), ('bundle-b', True)]   # B disabled A
AFTER A syncs again: [('bundle-a', True),  ('bundle-b', False)]  # A disabled B

What you think should happen instead?

A dag-processor that only owns a subset of the bundle configuration must not deactivate bundles it does not know about. Deactivation of genuinely-removed bundles should only happen when the process sees the full configuration.

Proposed fix: add a deactivate_missing flag to sync_bundles_to_db() (default True, preserving current single-processor behavior), and have DagFileProcessorManager.sync_bundles() pass deactivate_missing=False when the processor was started with a bundle filter (bundle_names_to_parse, i.e. --bundle-name). A PR implementing this follows.

Related: #69698 addresses a different over-eager-deactivation path in the same method (transient errors during bundle template extraction). This issue is about partial per-processor configuration, not error handling.

Operating System

Linux (Kubernetes; also reproduced locally on macOS)

Versions of Apache Airflow Providers

N/A (airflow-core)

Deployment

Other 3rd-party Helm chart

Deployment details

Multiple dag-processor deployments, each with its own AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST containing only the bundle that processor is responsible for, and started with --bundle-name to restrict parsing.

Anything else?

The affected code path (sync_bundles_to_db) is identical on main (3.3.0/3.4.0).

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct