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 <[email protected]> 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."""
