This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 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 0e516c37571 Improve deadline diagnostics for null DagRun fields
(#71767)
0e516c37571 is described below
commit 0e516c37571562919983e6faa9ca748e7fad3ae0
Author: Vincent Hsiao <[email protected]>
AuthorDate: Wed Sep 9 22:39:02 2026 +0800
Improve deadline diagnostics for null DagRun fields (#71767)
* Improve deadline diagnostics for null DagRun fields
* Limit null deadline warnings to DagRun timestamp references
* Add bugfix newsfragment for deadline diagnostics
* Fix deadline diagnostics newsfragment formatting
* Fix deadline diagnostics for null DagRun timestamps
---
airflow-core/newsfragments/71767.bugfix.rst | 1 +
.../src/airflow/serialization/definitions/dag.py | 16 ++++++
.../airflow/serialization/definitions/deadline.py | 9 ++--
airflow-core/tests/unit/models/test_dagrun.py | 58 ++++++++++++++++++++++
airflow-core/tests/unit/models/test_deadline.py | 35 ++++++++++++-
5 files changed, 115 insertions(+), 4 deletions(-)
diff --git a/airflow-core/newsfragments/71767.bugfix.rst
b/airflow-core/newsfragments/71767.bugfix.rst
new file mode 100644
index 00000000000..18c2de4a84a
--- /dev/null
+++ b/airflow-core/newsfragments/71767.bugfix.rst
@@ -0,0 +1 @@
+Fix misleading deadline warnings when DagRun-based deadline references
evaluate to None because the referenced timestamp field is null.
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py
b/airflow-core/src/airflow/serialization/definitions/dag.py
index 5b3a5352430..5d5c4763622 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -83,6 +83,11 @@ if TYPE_CHECKING:
log = structlog.get_logger(__name__)
+_DAGRUN_REFERENCE_REQUIRED_COLUMNS = {
+ SerializedReferenceModels.DagRunLogicalDateDeadline: "logical_date",
+ SerializedReferenceModels.DagRunQueuedAtDeadline: "queued_at",
+}
+
# TODO (GH-52141): Share definition with SDK?
class EdgeInfoType(TypedDict):
@@ -789,6 +794,17 @@ class SerializedDAG:
"deadline_alerts.deadline_created",
tags=prune_dict({"dag_id": self.dag_id, "team_name":
team_name}),
)
+ elif required_dagrun_column :=
_DAGRUN_REFERENCE_REQUIRED_COLUMNS.get(
+ type(deserialized_deadline_alert.reference)
+ ):
+ log.warning(
+ "skipping deadline alert because the deadline
reference evaluated to None",
+ dag_id=self.dag_id,
+ run_id=orm_dagrun.run_id,
+ deadline_alert_id=deadline_alert.id,
+
reference_type=deserialized_deadline_alert.reference.reference_name,
+ required_dagrun_column=required_dagrun_column,
+ )
@provide_session
def set_task_instance_state(
diff --git a/airflow-core/src/airflow/serialization/definitions/deadline.py
b/airflow-core/src/airflow/serialization/definitions/deadline.py
index e5108c6267c..5f462c63d1a 100644
--- a/airflow-core/src/airflow/serialization/definitions/deadline.py
+++ b/airflow-core/src/airflow/serialization/definitions/deadline.py
@@ -373,10 +373,13 @@ def _fetch_from_db(column, *, session: Session, dag_id:
str, run_id: str) -> dat
"""
from airflow.models import DagRun
- result = session.execute(select(column).where(DagRun.dag_id == dag_id,
DagRun.run_id == run_id)).scalar()
- if result is None:
+ row = session.execute(
+ select(column).where(DagRun.dag_id == dag_id, DagRun.run_id == run_id)
+ ).one_or_none()
+ if row is None:
logger.warning("Could not find DagRun for dag_id=%s, run_id=%s",
dag_id, run_id)
- return result
+ return None
+ return row[0]
@attrs.define(frozen=True)
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index fbcb9cb8cef..0a3250c3c62 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -1546,6 +1546,64 @@ class TestDagRun:
deadline = session.execute(select(Deadline)).scalars().one_or_none()
assert deadline.deadline_time == first_deadline_time
+ def test_dagrun_deadline_logs_when_reference_column_is_null(self, session,
deadline_test_dag, caplog):
+ scheduler_dag = deadline_test_dag(
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
+ interval=datetime.timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+
+ with caplog.at_level("WARNING"):
+ scheduler_dag.create_dagrun(
+ run_id="manual__null_logical_date",
+ run_type=DagRunType.MANUAL,
+ logical_date=None,
+ data_interval=None,
+ run_after=timezone.utcnow(),
+ start_date=timezone.utcnow(),
+ state=DagRunState.QUEUED,
+ triggered_by=DagRunTriggeredByType.TEST,
+ session=session,
+ )
+
+ assert session.execute(select(Deadline)).scalars().one_or_none() is
None
+ assert {
+ "event": "skipping deadline alert because the deadline reference
evaluated to None",
+ "dag_id": "test_dag",
+ "run_id": "manual__null_logical_date",
+ "reference_type": "DagRunLogicalDateDeadline",
+ "required_dagrun_column": "logical_date",
+ "log_level": "warning",
+ } in caplog
+ assert not any("Could not find DagRun" in record.message for record in
caplog.records)
+
+ def test_dagrun_deadline_does_not_warn_for_average_runtime_without_history(
+ self, session, deadline_test_dag, caplog
+ ):
+ scheduler_dag = deadline_test_dag(
+ deadline=DeadlineAlert(
+ reference=DeadlineReference.AVERAGE_RUNTIME(max_runs=10,
min_runs=5),
+ interval=datetime.timedelta(minutes=5),
+ callback=AsyncCallback(empty_callback_for_deadline),
+ ),
+ )
+
+ with caplog.at_level("WARNING",
logger="airflow.serialization.definitions.dag"):
+ self.create_dag_run(
+ dag=scheduler_dag,
+ logical_date=DEFAULT_DATE,
+ session=session,
+ )
+
+ assert session.execute(select(Deadline)).scalars().one_or_none() is
None
+ assert {
+ "event": "skipping deadline alert because the deadline reference
evaluated to None",
+ "reference_type": "AverageRuntimeDeadline",
+ "log_level": "warning",
+ } not in caplog
+
@mock.patch.object(Deadline, "prune_deadlines")
def test_dagrun_deadline_variable_interval_missing_variable_fails(self, _,
session, deadline_test_dag):
diff --git a/airflow-core/tests/unit/models/test_deadline.py
b/airflow-core/tests/unit/models/test_deadline.py
index 1771269b548..87c383d3e9c 100644
--- a/airflow-core/tests/unit/models/test_deadline.py
+++ b/airflow-core/tests/unit/models/test_deadline.py
@@ -42,7 +42,10 @@ from airflow.sdk.definitions.deadline import (
FixedDatetimeDeadline,
deadline_reference,
)
-from airflow.serialization.definitions.deadline import
SerializedReferenceModels
+from airflow.serialization.definitions.deadline import (
+ SerializedReferenceModels,
+ _fetch_from_db as _fetch_serialized_deadline_from_db,
+)
from airflow.utils.state import DagRunState
from tests_common.test_utils import db
@@ -468,6 +471,36 @@ class TestCalculatedDeadlineDatabaseCalls:
mock_fetch.assert_not_called()
assert result == DEFAULT_DATE + interval
+ def
test_serialized_fetch_from_db_returns_null_column_without_missing_dagrun_warning(
+ self, session, dag_maker, caplog
+ ):
+ with dag_maker(DAG_ID):
+ EmptyOperator(task_id="test_task")
+
+ dag_maker.create_dagrun(
+ logical_date=None,
+ run_id="manual__null_logical_date",
+ state=DagRunState.QUEUED,
+ )
+ session.commit()
+
+ with caplog.at_level("WARNING",
logger="airflow.serialization.definitions.deadline"):
+ result = _fetch_serialized_deadline_from_db(
+ DagRun.logical_date, session=session, dag_id=DAG_ID,
run_id="manual__null_logical_date"
+ )
+
+ assert result is None
+ assert not any("Could not find DagRun" in record.message for record in
caplog.records)
+
+ def test_serialized_fetch_from_db_logs_missing_dagrun(self, session,
caplog):
+ with caplog.at_level("WARNING",
logger="airflow.serialization.definitions.deadline"):
+ result = _fetch_serialized_deadline_from_db(
+ DagRun.logical_date, session=session, dag_id=DAG_ID,
run_id="missing_run"
+ )
+
+ assert result is None
+ assert any("Could not find DagRun" in record.message for record in
caplog.records)
+
def test_average_runtime_with_sufficient_history(self, session, dag_maker):
"""Test AverageRuntimeDeadline when enough historical data exists."""
with dag_maker(DAG_ID):
