Skip to content

Commit 7c77f8b

Browse files
[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 038ece6) Generated-by: Claude Code following the guidelines Co-authored-by: Sebastián Ortega <sebastian.ortegatorres@datadoghq.com>
1 parent b21256b commit 7c77f8b

4 files changed

Lines changed: 87 additions & 2 deletions

File tree

‎airflow-core/src/airflow/dag_processing/bundles/manager.py‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -271,7 +271,7 @@ def parse_config(self) -> None:
271271
self.log.info("DAG bundles loaded: %s", ", ".join(self._bundle_config.keys()))
272272

273273
@provide_session
274-
def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
274+
def sync_bundles_to_db(self, *, deactivate_missing: bool = True, session: Session = NEW_SESSION) -> None:
275275
"""
276276
Persist the configured DAG bundles into ``DagBundleModel`` rows.
277277
@@ -280,6 +280,14 @@ def sync_bundles_to_db(self, *, session: Session = NEW_SESSION) -> None:
280280
``DagModel`` / ``SerializedDagModel`` rows is the responsibility of
281281
``DagBag`` plus ``sync_bag_to_db`` (or, in production, the DAG
282282
processor); calling this method does not trigger that work.
283+
284+
:param deactivate_missing: When ``True`` (the default), any bundle stored in
285+
the database that is not present in this manager's config is marked
286+
inactive. This is only correct when the calling process sees the
287+
*complete* bundle configuration. When multiple dag-processors each run
288+
with a partial config (e.g. one bundle per processor), pass ``False`` so a
289+
processor does not disable bundles owned by other processors -- otherwise
290+
the processors repeatedly deactivate each other and never converge.
283291
"""
284292
self.log.debug("Syncing DAG bundles to the database")
285293

@@ -356,6 +364,11 @@ def _extract_and_sign_template(bundle_name: str) -> tuple[str | None, dict]:
356364
)
357365
bundle.teams = []
358366

367+
if not deactivate_missing:
368+
# This process only owns a subset of the bundles, so the remaining stored
369+
# bundles may well be owned by another dag-processor. Leave them alone.
370+
return
371+
359372
# Import here to avoid circular import
360373
from airflow.models.errors import ParseImportError
361374

‎airflow-core/src/airflow/dag_processing/manager.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -333,7 +333,9 @@ def _exit_gracefully(self, signum, frame):
333333

334334
def sync_bundles(self) -> None:
335335
"""Sync configured DAG bundles to the metadata database."""
336-
DagBundlesManager().sync_bundles_to_db()
336+
# When this processor only parses a subset of bundles, it does not see the full
337+
# bundle configuration and must not deactivate bundles owned by other processors.
338+
DagBundlesManager().sync_bundles_to_db(deactivate_missing=not self.bundle_names_to_parse)
337339

338340
def get_all_bundles(self) -> list[BaseDagBundle]:
339341
"""Return configured DAG bundles filtered by ``bundle_names_to_parse`` if provided."""

‎airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,14 @@ def path(self):
119119
}
120120
]
121121

122+
OTHER_BUNDLE_CONFIG = [
123+
{
124+
"name": "other-test-bundle",
125+
"classpath": "unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
126+
"kwargs": {"refresh_interval": 1},
127+
}
128+
]
129+
122130

123131
def test_get_bundle():
124132
"""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):
217225
assert "Removing ownership of team 'None'" not in caplog.text
218226

219227

228+
@pytest.mark.db_test
229+
@conf_vars({("core", "LOAD_EXAMPLES"): "False"})
230+
def test_sync_bundles_to_db_partial_config_does_not_disable_other_bundles(clear_db, session):
231+
"""
232+
Multiple dag-processors, each configured with only its own bundle, must not
233+
disable each other's bundles.
234+
235+
Each processor is started with ``--bundle-name`` (``bundle_names_to_parse``)
236+
and a ``dag_bundle_config_list`` containing only the bundle it owns. When
237+
``sync_bundles_to_db`` is told it only owns a subset (``deactivate_missing=False``),
238+
it must leave bundles it does not know about untouched instead of marking them
239+
inactive. Otherwise processor A disables B's bundle, B disables A's, and neither
240+
ever makes progress.
241+
"""
242+
243+
def _get_bundle_names_and_active():
244+
return session.execute(
245+
select(DagBundleModel.name, DagBundleModel.active).order_by(DagBundleModel.name)
246+
).all()
247+
248+
# Processor A: only knows "my-test-bundle".
249+
with patch.dict(
250+
os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
251+
):
252+
DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
253+
assert _get_bundle_names_and_active() == [("my-test-bundle", True)]
254+
255+
# Processor B: only knows "other-test-bundle". It must not touch A's bundle.
256+
with patch.dict(
257+
os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(OTHER_BUNDLE_CONFIG)}
258+
):
259+
DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
260+
assert _get_bundle_names_and_active() == [
261+
("my-test-bundle", True),
262+
("other-test-bundle", True),
263+
]
264+
265+
# Processor A runs again: still must not disable B's bundle.
266+
with patch.dict(
267+
os.environ, {"AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST": json.dumps(BASIC_BUNDLE_CONFIG)}
268+
):
269+
DagBundlesManager().sync_bundles_to_db(deactivate_missing=False)
270+
assert _get_bundle_names_and_active() == [
271+
("my-test-bundle", True),
272+
("other-test-bundle", True),
273+
]
274+
275+
220276
@conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(BASIC_BUNDLE_CONFIG)})
221277
@pytest.mark.parametrize("version", [None, "hello"])
222278
def test_view_url(version):

‎airflow-core/tests/unit/dag_processing/test_manager.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -295,6 +295,20 @@ def test_get_observed_filelocs_expands_zip_inner_paths(self, tmp_path):
295295
"test_zip.zip/broken_dag.py",
296296
}
297297

298+
def test_sync_bundles_deactivates_missing_when_owning_all_bundles(self):
299+
"""A processor with no bundle filter owns the full config and may deactivate missing bundles."""
300+
manager = DagFileProcessorManager(max_runs=1)
301+
with mock.patch("airflow.dag_processing.manager.DagBundlesManager") as mock_bundles_manager:
302+
manager.sync_bundles()
303+
mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=True)
304+
305+
def test_sync_bundles_does_not_deactivate_missing_when_filtered(self):
306+
"""A processor started with ``--bundle-name`` owns a subset and must not deactivate others."""
307+
manager = DagFileProcessorManager(max_runs=1, bundle_names_to_parse=["only-mine"])
308+
with mock.patch("airflow.dag_processing.manager.DagBundlesManager") as mock_bundles_manager:
309+
manager.sync_bundles()
310+
mock_bundles_manager.return_value.sync_bundles_to_db.assert_called_once_with(deactivate_missing=False)
311+
298312
@pytest.mark.usefixtures("clear_parse_import_errors")
299313
def test_refresh_dag_bundles_keeps_zip_inner_file_errors(self, session, tmp_path, configure_dag_bundles):
300314
bundle_path = tmp_path / "bundleone"

0 commit comments

Comments
 (0)