AIP-72: Supporting Pulling multiple XCOM values

[amoghrajesh]

closes: #45243

We can just handle xcoms for multiple task ids in the client directly by reusing the same task runner machinery.

DAG:

from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator

def push_to_xcom(**kwargs):
    value = "Hello, XCom!"
    return value

def push_to_xcom2(**kwargs):
    value = "Hello, XCom2!"
    return value

def pull_from_xcom(**kwargs):
    ti = kwargs['ti']
    xcom_value = ti.xcom_pull(task_ids=["push_xcom_task", "push_xcom_task2"])
    print(f"Retrieved XCom Value: {xcom_value}")


with DAG(
    'xcom_example',
    schedule=None,
    catchup=False,
) as dag:

    push_xcom_task = PythonOperator(
        task_id='push_xcom_task',
        python_callable=push_to_xcom,
    )

    push_xcom_task2 = PythonOperator(
        task_id='push_xcom_task2',
        python_callable=push_to_xcom2,
    )

    pull_xcom_task = PythonOperator(
        task_id='pull_xcom_task',
        python_callable=pull_from_xcom,
    )

    push_xcom_task >> push_xcom_task2 >> pull_xcom_task

Task1:

Task2:

Task3: (gets xcoms pushed by 1 and 2)

---

^ 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 newsfragments.

[amoghrajesh]

Not too sure where to add the tests for this.

[ashb]

How/where to test: add some tests that create a RunteimTaskInstance method and call the functions on it.

[kaxil]

Approving, but you should take a look too :D since I added the test -- to verify that you are happy with it.

[kaxil]

@amoghrajesh I merged this to tick off items :) -- but feel free to change anything in follow-up

[amoghrajesh]

Thanks @kaxil.
Yes it looks good.