This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 7c77f8be7f3 [v3-3-test] Don't deactivate DAG bundles owned by other
dag-processors (#69964) (#70017)
7c77f8be7f3 is described below
commit 7c77f8be7f36fb909367e7c84c86c439fcb6eb3c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Jul 19 23:51:33 2026 +0200
[v3-3-test] Don't deactivate DAG bundles owned by other dag-processors
(#69964) (#70017)
`DagBundlesManager.sync_bundles_to_db()` marks every stored bundle that is
not in the calling process's config as inactive. This assumes the caller
sees the complete bundle configuration.
When multiple dag-processors are each configured with a partial config (one
bundle per processor, a natural way to shard parsing), each processor treats
the other processors' bundles as "no longer found in config" and disables
them. Processor A disables B's bundle, B disables A's, and they flip
`dag_bundle.active` on every parse cycle -- the deployment never converges
and continuously logs "DAG bundle ... is no longer found in config and has
been disabled" for bundles that are actively configured elsewhere.
Add a `deactivate_missing` flag (default `True`, preserving existing
single-processor behavior) and have `DagFileProcessorManager.sync_bundles()`
pass `deactivate_missing=False` when the processor was started with a bundle
filter (`--bundle-name` / `bundle_names_to_parse`), i.e. when it only owns a
subset of the bundles.
closes: #69963
related: #69698
(cherry picked from commit 038ece6acf12794b2ae8ec84349a31e9cb629a4c)
Generated-by: Claude Code following the guidelines
Co-authored-by: Sebastián Ortega <[email protected]>
---
.../src/airflow/dag_processing/bundles/manager.py | 15 +++++-
airflow-core/src/airflow/dag_processing/manager.py | 4 +-
.../bundles/test_dag_bundle_manager.py | 56 ++++++++++++++++++++++
.../tests/unit/dag_processing/test_manager.py | 14 ++++++
4 files changed, 87 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/bundles/manager.py
b/airflow-core/src/airflow/dag_processing/bundles/manager.py
index 78c54266eda..2a26f224ae1 100644
--- a/airflow-core/src/airflow/dag_processing/bundles/manager.py
+++ b/airflow-core/src/airflow/dag_processing/bundles/manager.py
@@ -271,7 +271,7 @@ class DagBundlesManager(LoggingMixin):
self.log.info("DAG bundles loaded: %s", ",
".join(self._bundle_config.keys()))
@provide_session
- def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
+ def sync_bundles_to_db(self, *, deactivate_missing: bool = True, session:
Session = NEW_SESSION) -> None:
"""
Persist the configured DAG bundles into ``DagBundleModel`` rows.
@@ -280,6 +280,14 @@ class DagBundlesManager(LoggingMixin):
``DagModel`` / ``SerializedDagModel`` rows is the responsibility of
``DagBag`` plus ``sync_bag_to_db`` (or, in production, the DAG
processor); calling this method does not trigger that work.
+
+ :param deactivate_missing: When ``True`` (the default), any bundle
stored in
+ the database that is not present in this manager's config is marked
+ inactive. This is only correct when the calling process sees the
+ *complete* bundle configuration. When multiple dag-processors each
run
+ with a partial config (e.g. one bundle per processor), pass
``False`` so a
+ processor does not disable bundles owned by other processors --
otherwise
+ the processors repeatedly deactivate each other and never converge.
"""
self.log.debug("Syncing DAG bundles to the database")
@@ -356,6 +364,11 @@ class DagBundlesManager(LoggingMixin):
)
bundle.teams = []
+ if not deactivate_missing:
+ # This process only owns a subset of the bundles, so the remaining
stored
+ # bundles may well be owned by another dag-processor. Leave them
alone.
+ return
+
# Import here to avoid circular import
from airflow.models.errors import ParseImportError
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index b2428985a65..02028f1bb5c 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -333,7 +333,9 @@ class DagFileProcessorManager(LoggingMixin):
def sync_bundles(self) -> None:
"""Sync configured DAG bundles to the metadata database."""
- DagBundlesManager().sync_bundles_to_db()
+ # When this processor only parses a subset of bundles, it does not see
the full
+ # bundle configuration and must not deactivate bundles owned by other
processors.
+ DagBundlesManager().sync_bundles_to_db(deactivate_missing=not
self.bundle_names_to_parse)
def get_all_bundles(self) -> list[BaseDagBundle]:
"""Return configured DAG bundles filtered by ``bundle_names_to_parse``
if provided."""
diff --git
a/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
b/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
index 0d17069c831..4b3fc1078ab 100644
--- a/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
+++ b/airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py
@@ -119,6 +119,14 @@ BASIC_BUNDLE_CONFIG = [
}
]
+OTHER_BUNDLE_CONFIG = [
+ {
+ "name": "other-test-bundle",
+ "classpath":
"unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+ "kwargs": {"refresh_interval": 1},
+ }
+]
+
def test_get_bundle():
"""Test that get_bundle builds and returns a bundle."""
@@ -217,6 +225,54 @@ def
test_sync_bundles_to_db_does_not_log_removing_none_team(clear_db, caplog):
assert "Removing ownership of team 'None'" not in caplog.text
[email protected]_test
+@conf_vars({("core", "LOAD_EXAMPLES"): "False"})
+def
test_sync_bundles_to_db_partial_config_does_not_disable_other_bundles(clear_db,
session):
+ """
+ Multiple dag-processors, each configured with only its own bundle, must not
+ disable each other's bundles.
+
+ Each processor is started with ``--bundle-name``
(``bundle_names_to_parse``)
+ and a ``dag_bundle_config_list`` containing only the bundle it owns. When
+ ``sync_bundles_to_db`` is told it only owns a subset
(``deactivate_missing=False``),
+ it must leave bundles it does not know about untouched instead of marking
them
+ inactive. Otherwise processor A disables B's bundle, B disables A's, and
neither
+ ever makes progress.
+ """
+
+ def _get_bundle_names_and_active():
+ return session.execute(
+ select(DagBundleModel.name,
DagBundleModel.active).order_by(DagBundleModel.name)
+ ).all()
+
+ # Processor A: only knows "my-test-bundle".
+ with patch.dict(
+ os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST":
json.dumps(BASIC_BUNDLE_CONFIG)}
+ ):
+ DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
+ assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
+
+ # Processor B: only knows "other-test-bundle". It must not touch A's
bundle.
+ with patch.dict(
+ os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST":
json.dumps(OTHER_BUNDLE_CONFIG)}
+ ):
+ DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
+ assert _get_bundle_names_and_active() == [
+ ("my-test-bundle", True),
+ ("other-test-bundle", True),
+ ]
+
+ # Processor A runs again: still must not disable B's bundle.
+ with patch.dict(
+ os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST":
json.dumps(BASIC_BUNDLE_CONFIG)}
+ ):
+ DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
+ assert _get_bundle_names_and_active() == [
+ ("my-test-bundle", True),
+ ("other-test-bundle", True),
+ ]
+
+
@conf_vars({("dag_processor", "dag_bundle_config_list"):
json.dumps(BASIC_BUNDLE_CONFIG)})
@pytest.mark.parametrize("version", [None, "hello"])
def test_view_url(version):
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index 4b55b6b2f1e..d7f50e6335c 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -295,6 +295,20 @@ class TestDagFileProcessorManager:
"test_zip.zip/broken_dag.py",
}
+ def test_sync_bundles_deactivates_missing_when_owning_all_bundles(self):
+ """A processor with no bundle filter owns the full config and may
deactivate missing bundles."""
+ manager = DagFileProcessorManager(max_runs=1)
+ with mock.patch("airflow.dag_processing.manager.DagBundlesManager") as
mock_bundles_manager:
+ manager.sync_bundles()
+
mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=True)
+
+ def test_sync_bundles_does_not_deactivate_missing_when_filtered(self):
+ """A processor started with ``--bundle-name`` owns a subset and must
not deactivate others."""
+ manager = DagFileProcessorManager(max_runs=1,
bundle_names_to_parse=["only-mine"])
+ with mock.patch("airflow.dag_processing.manager.DagBundlesManager") as
mock_bundles_manager:
+ manager.sync_bundles()
+
mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=False)
+
@pytest.mark.usefixtures("clear_parse_import_errors")
def test_refresh_dag_bundles_keeps_zip_inner_file_errors(self, session,
tmp_path, configure_dag_bundles):
bundle_path = tmp_path / "bundleone"
