This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 14551facad6 Fix Dag callbacks silently dropped when version inflation
check blocks parsing (#70987)
14551facad6 is described below
commit 14551facad691258276081313485f3386dfca377
Author: Ephraim Anierobi <[email protected]>
AuthorDate: Thu Aug 20 08:16:30 2026 +0100
Fix Dag callbacks silently dropped when version inflation check blocks
parsing (#70987)
---
.../src/airflow/dag_processing/processor.py | 7 +++-
.../tests/unit/dag_processing/test_processor.py | 42 ++++++++++++++++++++++
2 files changed, 48 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index a44f9f37b9e..3805de9fdf5 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -237,7 +237,12 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
stability_check_result = check_dag_file_stability(os.fspath(msg.file))
- if stability_check_error_dict :=
stability_check_result.get_error_format_dict(msg.file, msg.bundle_path):
+ # Callback runs must not be blocked by the stability check: callbacks for
+ # already-scheduled runs still have to execute, and they never produce a
+ # parsing result anyway.
+ if not msg.callback_requests and (
+ stability_check_error_dict :=
stability_check_result.get_error_format_dict(msg.file, msg.bundle_path)
+ ):
# If Dag stability check level is error, we shouldn't parse the Dags
and return the result early
return DagFileParsingResult(
fileloc=msg.file,
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index 2ced95916f7..f54b82fef82 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -681,6 +681,48 @@ def test_parse_file_static_check_with_error():
assert "Don't use the variables as arguments" in
next(iter(result.import_errors.values()))
+@conf_vars({("dag_processor", "dag_version_inflation_check_level"): "error"})
+def test_parse_file_static_check_error_still_executes_callbacks(spy_agency):
+ """Callbacks for already-scheduled runs must run even when the stability
check blocks parsing."""
+ from airflow import DAG
+
+ called = False
+
+ def on_failure(context):
+ nonlocal called
+ called = True
+
+ dag = DAG(dag_id="a", on_failure_callback=on_failure)
+
+ def fake_collect_dags(self, *args, **kwargs):
+ self.dags[dag.dag_id] = dag
+
+ spy_agency.spy_on(DagBag.collect_dags, call_fake=fake_collect_dags,
owner=DagBag)
+
+ requests = [
+ DagCallbackRequest(
+ filepath="test_dag_version_inflation_check.py",
+ msg="Message",
+ dag_id="a",
+ run_id="b",
+ bundle_name="testing",
+ bundle_version=None,
+ )
+ ]
+ result = _parse_file(
+ DagFileParseRequest(
+ file=f"{TEST_DAG_FOLDER}/test_dag_version_inflation_check.py",
+ bundle_path=TEST_DAG_FOLDER,
+ bundle_name="testing",
+ callback_requests=requests,
+ ),
+ log=structlog.get_logger(),
+ )
+
+ assert result is None
+ assert called is True
+
+
def test_parse_file_static_check_with_default_warning():
result = _parse_file(
DagFileParseRequest(
