Fix new DAG versions from serializing a retry_policy object

[vatsrahul1001]

What's wrong

A mapped task that uses a retry policy (.partial(retry_policy=…).expand(…) or
@task(retry_policy=…).expand(…)) gets a new DAG version created on nearly every
DAG-processor re-serialization (~30s) — even when the DAG is idle and never run —
spamming the DAG version history and bloating the metadata DB (dag_version /
serialized_dag).

Root cause

A mapped operator keeps retry_policy in partial_kwargs, and the mapped-operator
serializer has no serializer for a RetryPolicy, so it falls back to str(obj) —
"<…ExceptionRetryPolicy object at 0x…>". That memory address changes on every parse,
so dag_hash changes each time, defeating write_dag's "unchanged" check and creating
a new DagVersion on every re-serialization. Non-mapped operators already avoid this by
excluding the object and storing only a has_retry_policy flag.

Fix

Mirror the non-mapped path in the mapped-operator serializer: don't serialize the object,
store only a has_retry_policy flag, and add has_retry_policy to SerializedMappedOperator
so it round-trips. Runtime is unaffected — the worker re-parses the DAG source for the live
policy, so retries still work.

Testing

Adds regression tests that a mapped task with a retry_policy (a) serializes
deterministically (identical across parses, no embedded object) and (b) round-trips
has_retry_policy=True. Both fail on main and pass with the fix.

Before

1. Try below DAG and create a dag_run.
2. Execute SELECT count(*) FROM dag_version WHERE dag_id='aip105_mapped_serialize_repro' and confirm new version is created

from airflow.sdk import DAG, ExceptionRetryPolicy, RetryAction, RetryRule, task

POLICY = ExceptionRetryPolicy(
    rules=[
        RetryRule(
            exception=ConnectionError,
            action=RetryAction.RETRY,
            retry_delay=timedelta(seconds=30),
            reason="Transient, backing off 30s",
        ),
    ],
)

with DAG(
    dag_id="aip105_mapped_serialize_repro",
    schedule=None,
    catchup=False,
    tags=["qa", "aip105", "serialization-bug"],
):

    @task(retries=3, retry_policy=POLICY)
    def mapped_with_policy(x):
        return x

    @task(retries=3, retry_policy=POLICY)
    def plain_with_policy():
        return 1

    mapped_with_policy.expand(x=[1, 2, 3])
    plain_with_policy()

After

1. Run same DAG and check new version is not getting created

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

[kaxil]

Few non-blocking comments but lgtm otherwise

[github-actions]

Backport successfully created: v3-3-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-3-test