AIP-72: Gracefully handle 'not-found' XCOMs in task sdk API client

[amoghrajesh]

closes: #45341

Testing results

DAG used:

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

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


def pull_from_xcom(**kwargs):
    ti = kwargs['ti']
    xcom_value = ti.xcom_pull(task_ids='invalid_id')
    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,
    )

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

    push_xcom_task >> pull_xcom_task

Legacy

Task 1 XCom pushed:

Task 2 XCom consumed as None when absent:

Task SDK

Task 1 XCom pushed:

Task 2 XCom consumed as None when absent:

---

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

[jscheffl]

Looks good, just one comment. But not critical to move ahead.

[jscheffl]

Oh, much simpler now :-D

[potiuk]

@amoghrajesh - can you please rebase that one -> we found and issue with @jscheffl with the new caching scheme - fixed in #45347 that would run "main" version of the tests. I am asking in all affected PRs to rebase.

[amoghrajesh]

@uranusjr @jscheffl its already handled using an event hook

[amoghrajesh]

@amoghrajesh - can you please rebase that one -> we found and issue with @jscheffl with the new caching scheme - fixed in #45347 that would run "main" version of the tests. I am asking in all affected PRs to rebase.

Oh ok, let me rebase this one

[amoghrajesh]

Hi @uranusjr are you ok with the reply on this comment? #45344 (comment)