Fix XCom deserialization for mapped tasks with custom backend

[amoghrajesh]

closes: #50636

What the problem is?

While using a dag that has tasks consuming xcoms from upstream tasks which are mapped and a custom xcom backend is configured, the downstream tasks would not get the "data" of the xcom or the value of the xcom, but would instead get the reference of it.

In case of object store backend, the downstream tasks were suffering with the "path" on the store of the xcom.

Problem

Due to some recent refactoring done by: #50011, the deserialize function on the custom xcom backend was not consumed and "BaseXcom" was consumed always leading to it doing the most basic deserialize -- (serde deser) and hence the retrurned object remained equal to the reference of it.

Testing

DAG used (same as the reported one)

from airflow.decorators import dag, task
from airflow.models.xcom import XCom

@task
def generate_data():
    print("XCom.__name__ --->", XCom.__name__)
    return [1, 2, 3]


@task
def filter_item_from_mapped_task(item):
    print("filter_item_from_mapped_task item --->", item)
    if item==1:
        return None
    return item


@task
def print_item(item):
    print("print_item item --->", item)


@dag(
    dag_id='test_bug_dag',
)
def test_bug_dag():
    data = generate_data()
    filtered_data = filter_item_from_mapped_task.expand(item=data)
    print_item.expand(item=filtered_data)

test_bug_dag()

generate_data pushes right stuff:

filter_item_from_mapped_task pushes right stuff:

map index 0:

map index 1:

map index 2:

Push on s3 is as expected:

Consumed rightly by the print_item task now:

map index 0:

map index 1:

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Good find

[github-actions]

Backport successfully created: v3-0-test

Status	Branch	Result
✅	v3-0-test