You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ra...@apache.org on 2026/07/02 10:05:15 UTC
(airflow) 01/01: Fix non-deterministic serialization of mapped task retry_policy
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch fix-mapped-retry-policy-serialization
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 50d575b7a58a586633dd93e2fdeaeda2214d6773
Author: Rahul Vats <ra...@gmail.com>
AuthorDate: Thu Jul 2 15:34:45 2026 +0530
Fix non-deterministic serialization of mapped task retry_policy
A mapped operator keeps retry_policy in partial_kwargs, but the mapped-operator
serializer has no serializer for a RetryPolicy, so it fell back to str(obj) --
embedding the object's per-process memory address. That made the serialized DAG
non-deterministic: dag_hash changed on every parse, defeating write_dag's
"unchanged" check and creating a new DagVersion on nearly every DAG-processor
re-serialization (even for idle, never-run DAGs), polluting version history and
bloating the metadata DB.
Non-mapped operators already avoid this by excluding the object and storing only
a has_retry_policy flag; this makes the mapped path do the same. Runtime is
unaffected -- the worker re-parses the DAG source for the live policy.
---
.../serialization/definitions/mappedoperator.py | 4 ++
.../airflow/serialization/serialized_objects.py | 9 ++++
.../unit/serialization/test_serialized_objects.py | 53 ++++++++++++++++++++++
3 files changed, 66 insertions(+)
diff --git a/airflow-core/src/airflow/serialization/definitions/mappedoperator.py b/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
index 1cf6d357e65..348e8423100 100644
--- a/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
+++ b/airflow-core/src/airflow/serialization/definitions/mappedoperator.py
@@ -262,6 +262,10 @@ class SerializedMappedOperator(DAGNode):
def has_on_skipped_callback(self) -> bool:
return self._get_partial_kwargs_or_operator_default("has_on_skipped_callback")
+ @property
+ def has_retry_policy(self) -> bool:
+ return self._get_partial_kwargs_or_operator_default("has_retry_policy")
+
@property
def run_as_user(self) -> str | None:
return self._get_partial_kwargs_or_operator_default("run_as_user")
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py
index 4ba0bbefdce..70c7ae8e008 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -974,6 +974,15 @@ class OperatorSerialization(DAGNode, BaseSerialization):
if cls._is_excluded(v, k, op):
continue
+ if k == "retry_policy":
+ # A RetryPolicy has no serializer, so serializing it falls back to
+ # str(obj) -- which embeds the object's memory address and makes the
+ # serialized DAG non-deterministic (a new DagVersion every parse). Mirror
+ # the non-mapped path and keep only a has_retry_policy flag; the worker
+ # re-parses the DAG source for the live policy.
+ if bool(v):
+ serialized_op["partial_kwargs"]["has_retry_policy"] = True
+ continue
if k in _OPERATOR_CALLBACK_FIELDS:
if bool(v):
serialized_op["partial_kwargs"][f"has_{k}"] = True
diff --git a/airflow-core/tests/unit/serialization/test_serialized_objects.py b/airflow-core/tests/unit/serialization/test_serialized_objects.py
index 7de4c87148a..465b100ef19 100644
--- a/airflow-core/tests/unit/serialization/test_serialized_objects.py
+++ b/airflow-core/tests/unit/serialization/test_serialized_objects.py
@@ -1309,6 +1309,59 @@ class TestRetryPolicySerialization:
task = deserialized.task_dict["op_no_policy"]
assert task.has_retry_policy is False
+ def test_mapped_task_retry_policy_serializes_as_flag(self):
+ """A mapped task's retry_policy must serialize as has_retry_policy, not the object."""
+ import json
+
+ from airflow.sdk import DAG, task
+ from airflow.sdk.definitions.retry_policy import ExceptionRetryPolicy, RetryAction, RetryRule
+ from airflow.serialization.serialized_objects import DagSerialization
+
+ policy = ExceptionRetryPolicy(
+ rules=[RetryRule(exception=ValueError, action=RetryAction.FAIL, reason="bad data")],
+ )
+
+ with DAG(dag_id="test_mapped_retry_policy_ser", start_date=DEFAULT_DATE) as dag:
+
+ @task(retries=3, retry_policy=policy)
+ def mapped(x):
+ return x
+
+ mapped.expand(x=[1, 2, 3])
+
+ serialized = DagSerialization.serialize_dag(dag)
+ # The RetryPolicy object must never be embedded -- str(obj) leaks a memory address.
+ assert "ExceptionRetryPolicy object at 0x" not in json.dumps(serialized)
+
+ deserialized = DagSerialization.deserialize_dag(serialized)
+ assert deserialized.task_dict["mapped"].has_retry_policy is True
+
+ def test_mapped_task_retry_policy_serialization_is_deterministic(self):
+ """Serializing the same mapped-task-with-policy DAG twice yields identical output.
+
+ Regression test: the RetryPolicy object was serialized via str(obj), embedding a
+ per-process memory address, so every re-parse produced a different serialized DAG
+ (and a spurious new DagVersion).
+ """
+ from airflow.sdk import DAG, task
+ from airflow.sdk.definitions.retry_policy import ExceptionRetryPolicy, RetryAction, RetryRule
+ from airflow.serialization.serialized_objects import DagSerialization
+
+ def build():
+ policy = ExceptionRetryPolicy(
+ rules=[RetryRule(exception=ValueError, action=RetryAction.FAIL)],
+ )
+ with DAG(dag_id="test_mapped_retry_policy_determinism", start_date=DEFAULT_DATE) as dag:
+
+ @task(retries=3, retry_policy=policy)
+ def mapped(x):
+ return x
+
+ mapped.expand(x=[1, 2, 3])
+ return dag
+
+ assert DagSerialization.serialize_dag(build()) == DagSerialization.serialize_dag(build())
+
class TestKubernetesImportAvoidance:
"""Test that serialization doesn't import kubernetes unnecessarily."""