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 690a64dc2f4 [v3-3-test] Fix #60763: Deactivate legacy DAGs with NULL
bundle_name during upgrade from 2.x to 3.x (#61019) (#70662)
690a64dc2f4 is described below
commit 690a64dc2f42abee76aef870ecff172afebd7c42
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Jul 29 09:33:11 2026 +0200
[v3-3-test] Fix #60763: Deactivate legacy DAGs with NULL bundle_name during
upgrade from 2.x to 3.x (#61019) (#70662)
* Fix #60763: Deactivate legacy DAGs with NULL bundle_name during upgrade
from 2.x to 3.x
This uses a query-based approach to deactivate legacy DAGs with NULL
bundle_name, avoiding the need for a database migration that caused CI failures
and addressing maintainer feedback.
* Trigger CI re-run due to git clone flake
* Expand comment explaining why NULL bundle_name means stale
* Add unit test covering deactivation of DAGs with NULL bundle_name
* Drop stray local conftest from the merge
(cherry picked from commit e272ae7f977bbd6174d1a7103b1e79851ef07b03)
Co-authored-by: Pradeep Kalluri
<[email protected]>
---
airflow-core/src/airflow/dag_processing/manager.py | 12 +++++--
.../tests/unit/dag_processing/test_manager.py | 42 +++++++++++++++++++++-
2 files changed, 51 insertions(+), 3 deletions(-)
diff --git a/airflow-core/src/airflow/dag_processing/manager.py
b/airflow-core/src/airflow/dag_processing/manager.py
index c77f6d08a07..c936d3548f6 100644
--- a/airflow-core/src/airflow/dag_processing/manager.py
+++ b/airflow-core/src/airflow/dag_processing/manager.py
@@ -469,9 +469,17 @@ class DagFileProcessorManager(LoggingMixin):
for dag in dags_parsed:
# Dags whose bundle has been removed from config (bundle no longer
active) are stale —
# the processor has stopped parsing their files, so the time-based
check below would never fire.
- if dag.bundle_name in inactive_bundles:
+ #
+ # A NULL bundle_name means the row predates bundles (carried over
from Airflow 2.x) and has not
+ # been parsed since the upgrade — parsing is what fills
bundle_name in. If the file was removed
+ # as part of the upgrade, no parse will ever happen, so
bundle_name stays NULL forever. Such a
+ # row can never hit the time-based check below either, because
that matches on
+ # (bundle_name, relative_fileloc) and there is no bundle to match
against, so without this
+ # branch the Dag stays active in the UI indefinitely. If the file
does still exist, the next
+ # parse fills in bundle_name and clears is_stale, so a Dag
deactivated here is reactivated.
+ if dag.bundle_name is None or dag.bundle_name in inactive_bundles:
self.log.info(
- "Deactivating Dag %s. Its bundle %s is no longer active.",
+ "Deactivating Dag %s. Its bundle %s is no longer active or
is NULL.",
dag.dag_id,
dag.bundle_name,
)
diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py
b/airflow-core/tests/unit/dag_processing/test_manager.py
index f08aa815a72..23e1d97e6f7 100644
--- a/airflow-core/tests/unit/dag_processing/test_manager.py
+++ b/airflow-core/tests/unit/dag_processing/test_manager.py
@@ -28,7 +28,7 @@ import signal
import textwrap
import time
import zipfile
-from collections import OrderedDict, defaultdict
+from collections import OrderedDict, defaultdict, namedtuple
from datetime import datetime, timedelta
from pathlib import Path
from socket import socket, socketpair
@@ -1069,6 +1069,46 @@ class TestDagFileProcessorManager:
)
assert is_stale_by_dag == {"dag_in_inactive_bundle": True,
"dag_in_active_bundle": False}
+ @pytest.mark.usefixtures("testing_dag_bundle")
+ def test_deactivate_stale_dags_marks_dags_with_null_bundle_name(self,
session):
+ """Dags carried over from Airflow 2.x keep a NULL bundle_name and must
still be deactivated.
+
+ Their files were removed during the upgrade, so nothing will ever
parse them and fill the
+ column in, and the time-based check cannot reach them either (see
#60763).
+
+ Migration ``0082_3_1_0_make_bundle_name_not_nullable`` backfills the
column and makes it NOT
+ NULL, so the row can no longer be stored as NULL; the scan is fed the
row the way a database
+ upgraded from 2.x to 3.0.x still holds it.
+ """
+ session.add(
+ DagModel(
+ dag_id="legacy_dag",
+ bundle_name="testing",
+ relative_fileloc="legacy_file.py",
+ last_parsed_time=timezone.utcnow(),
+ is_stale=False,
+ )
+ )
+ session.flush()
+
+ LegacyRow = namedtuple("LegacyRow", "dag_id bundle_name fileloc
last_parsed_time relative_fileloc")
+ original_execute = session.execute
+
+ def execute_with_null_bundle_name(statement, *args, **kwargs):
+ result = original_execute(statement, *args, **kwargs)
+ if getattr(statement, "is_select", False) and "relative_fileloc"
in str(statement):
+ return [
+ LegacyRow(r.dag_id, None, r.fileloc, r.last_parsed_time,
r.relative_fileloc)
+ for r in result
+ ]
+ return result
+
+ manager = DagFileProcessorManager(max_runs=1, processor_timeout=10 *
60)
+ with mock.patch.object(session, "execute",
side_effect=execute_with_null_bundle_name):
+ manager.deactivate_stale_dags(last_parsed={}, session=session)
+
+ assert session.scalar(select(DagModel.is_stale).where(DagModel.dag_id
== "legacy_dag"))
+
@mock.patch("airflow.dag_processing.manager.is_lock_not_available_error")
@pytest.mark.usefixtures("testing_dag_bundle")
def test_deactivate_stale_dags_handles_lock_timeout(self,
mock_is_lock_not_available, session, caplog):
