This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new ad1dc8eafa6 [v3-4-test] Add `partition_key` to the `DagRun Finished`
log line (#76557) (#76564)
ad1dc8eafa6 is described below
commit ad1dc8eafa6878b08e91b777b98d930bc2d37d3e
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Nov 6 10:38:50 2026 +0200
[v3-4-test] Add `partition_key` to the `DagRun Finished` log line (#76557)
(#76564)
The scheduler logs "DagRun Finished: ..." when a Dag run reaches
success or failure, but the line did not say which partition the run
was for, so partitioned runs (AIP-76) could not be told apart in logs.
The Dag run span already carries airflow.dag_run.partition_key.
Append partition_key=%s as the last field, keeping every existing field
and the trailing comma unchanged so current parsers keep working. Runs
without a partition key log None, like the other optional fields.
(cherry picked from commit 77762c256df493e73d7b5e050b7e71ea85b74795)
Co-authored-by: Constance Martineau <[email protected]>
---
airflow-core/src/airflow/models/dagrun.py | 4 +++-
airflow-core/tests/unit/models/test_dagrun.py | 33 +++++++++++++++++++++++++++
2 files changed, 36 insertions(+), 1 deletion(-)
diff --git a/airflow-core/src/airflow/models/dagrun.py
b/airflow-core/src/airflow/models/dagrun.py
index d5b8d600ca4..2e1923b30a8 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -1447,7 +1447,8 @@ class DagRun(Base, LoggingMixin):
"DagRun Finished: dag_id=%s, logical_date=%s, run_id=%s, "
"run_start_date=%s, run_end_date=%s, run_duration=%s, "
"state=%s, run_type=%s, "
- "data_interval_start=%s, data_interval_end=%s,"
+ "data_interval_start=%s, data_interval_end=%s, "
+ "partition_key=%s,"
)
self.log.info(
msg,
@@ -1465,6 +1466,7 @@ class DagRun(Base, LoggingMixin):
self.run_type,
self.data_interval_start,
self.data_interval_end,
+ self.partition_key,
)
session.flush()
try:
diff --git a/airflow-core/tests/unit/models/test_dagrun.py
b/airflow-core/tests/unit/models/test_dagrun.py
index 7a63545926b..430103ecc05 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -639,6 +639,39 @@ class TestDagRun:
dag_run.update_state()
assert dag_run.state == DagRunState.SUCCESS
+ @pytest.mark.parametrize(
+ ("partition_key", "expected_suffix"),
+ [
+ pytest.param("2026-01-01T00:00:00", ",
partition_key=2026-01-01T00:00:00,", id="with-key"),
+ pytest.param(None, ", partition_key=None,", id="without-key"),
+ ],
+ )
+ def test_dagrun_finished_log_ends_with_partition_key(
+ self, dag_maker, session, caplog, partition_key, expected_suffix
+ ):
+ with dag_maker(
+ dag_id="test_dagrun_finished_log",
+ schedule=CronPartitionTimetable("0 0 * * *",
timezone=pendulum.UTC),
+ start_date=datetime.datetime(2026, 1, 1),
+ session=session,
+ ):
+ EmptyOperator(task_id="task")
+ dag_run = dag_maker.create_dagrun(
+ logical_date=None,
+ partition_date=datetime.datetime(2026, 1, 1, tzinfo=pendulum.UTC)
if partition_key else None,
+ partition_key=partition_key,
+ )
+ for ti in dag_run.get_task_instances(session=session):
+ ti.set_state(TaskInstanceState.SUCCESS, session=session)
+
+ with caplog.at_level("INFO"):
+ dag_run.update_state(session=session)
+
+ finished = [r.getMessage() for r in caplog.records if
r.getMessage().startswith("DagRun Finished:")]
+ assert len(finished) == 1
+ assert ", data_interval_end=" in finished[0]
+ assert finished[0].endswith(expected_suffix)
+
def
test_dagrun_not_stuck_in_running_when_all_tasks_instances_are_removed(self,
dag_maker, session):
"""
Tests that a DAG run succeeds when all tasks are removed
