This is an automated email from the ASF dual-hosted git repository.
ashb 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 55e341fc167 [v3-3-test] Fix Dag callbacks silently dropped when
version inflation check blocks parsing (#70987) (#71865)
55e341fc167 is described below
commit 55e341fc1670aab2bca4127153f672c77c28b884
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Aug 20 08:47:58 2026 +0100
[v3-3-test] Fix Dag callbacks silently dropped when version inflation check
blocks parsing (#70987) (#71865)
---
.../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 ed06b9d83d2..89b5243d762 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -235,7 +235,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 c48fe43c887..a9ee6feab07 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -637,6 +637,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(
