Airflow 3. xcom_pull different behavior between Airflow 2 and 3

[kosteev]

Apache Airflow version

3.0.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

With the following DAG:

from __future__ import annotations

import logging
from datetime import datetime

from airflow.decorators import task

from airflow.models.dag import DAG


log = logging.getLogger(__name__)


with DAG(
    "xcom_test",
    schedule="* * * * *",  # Override to match your needs,
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example", "sheets"],
) as dag:
  @task
  def xcom_string():
    return "string"

  @task
  def xcom_dict():
    return {"a": "b"}


  @task
  def xcom_list():
    return [("a", "b"), ("c", "d")]


  @task
  def read_xcom(**kwargs):
    xcom_string_task_value = kwargs["task_instance"].xcom_pull("xcom_string")
    print("xcom_string_task", type(xcom_string_task_value), xcom_string_task_value)

    xcom_string_task_value_ids = kwargs["task_instance"].xcom_pull(task_ids=["xcom_string"], key="return_value")
    print("xcom_string_task_ids", type(xcom_string_task_value_ids), xcom_string_task_value_ids, xcom_string_task_value_ids[0])

    xcom_dict_task_value = kwargs["task_instance"].xcom_pull("xcom_dict")
    print("xcom_dict_task", type(xcom_dict_task_value), xcom_dict_task_value)

    xcom_dict_task_value_ids = kwargs["task_instance"].xcom_pull(task_ids=["xcom_dict"], key="return_value")
    print("xcom_dict_task_ids", type(xcom_dict_task_value_ids), xcom_dict_task_value_ids, xcom_dict_task_value_ids[0])

    xcom_list_task_value = kwargs["task_instance"].xcom_pull("xcom_list")
    print("xcom_list_task", type(xcom_list_task_value), xcom_list_task_value)

    xcom_list_task_value_ids = kwargs["task_instance"].xcom_pull(task_ids=["xcom_list"], key="return_value")
    print("xcom_list_task_ids", type(xcom_list_task_value_ids), xcom_list_task_value_ids, xcom_list_task_value_ids[0])


  xcom_string = xcom_string()
  xcom_dict = xcom_dict()
  xcom_list = xcom_list()

  read_xcom_task = read_xcom()
  xcom_string >> xcom_dict >> xcom_list >> read_xcom_task

In Airflow 2, the output is:

[2025-04-22, 08:13:20 UTC] {logging_mixin.py:190} INFO - xcom_string_task <class 'str'> string
[2025-04-22, 08:13:20 UTC] {logging_mixin.py:190} INFO - xcom_string_task_ids <class 'airflow.models.xcom.LazyXComSelectSequence'> LazySelectSequence([1 item]) string
[2025-04-22, 08:13:20 UTC] {logging_mixin.py:190} INFO - xcom_dict_task <class 'dict'> {'a': 'b'}
[2025-04-22, 08:13:20 UTC] {logging_mixin.py:190} INFO - xcom_dict_task_ids <class 'airflow.models.xcom.LazyXComSelectSequence'> LazySelectSequence([1 item]) {'a': 'b'}
[2025-04-22, 08:13:20 UTC] {logging_mixin.py:190} INFO - xcom_list_task <class 'list'> [['a', 'b'], ['c', 'd']]
[2025-04-22, 08:13:20 UTC] {logging_mixin.py:190} INFO - xcom_list_task_ids <class 'airflow.models.xcom.LazyXComSelectSequence'> LazySelectSequence([1 item]) [['a', 'b'], ['c', 'd']]
[2025-04-22, 08:13:20 UTC] {python.py:240} INFO - Done. Returned value was: None

In Airflow 3:

What you think should happen instead?

The same behavior in Airflow 2 and 3.
For the case (task_ids=["{task_id}"]), the difference is that in Airflow 2 the result will be list of values, in Airflow 3 the result will be the value itself.

How to reproduce

See section above

Operating System

Linux

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[kosteev]

Related issue #46513

[amoghrajesh]

Hello @kosteev i have responded to you over email. But reiterating for others:

Yes, this change was intentional in Airflow 3. Here’s why:

In Airflow 2, the xcom_pull() method allowed pulling XComs by key without specifying task_ids, despite the fact that the underlying
DB model defines task_id as part of the XCom primary key. This created ambiguity: if two tasks pushed XComs with the same key,
xcom_pull() would pull whichever one happened to be first, leading to unpredictable behavior.

Airflow 3 resolves this inconsistency by requiring task_ids when pulling by key. This change aligns with the task-scoped nature of
XComs as defined by the schema, ensuring predictable and consistent behavior.

We will be fixing this in a later patch release but to answer your question, it is an intended change and is being handled in migration
guide via the PR: #49455

[amoghrajesh]

Related comment: 1c6509e#diff-297ac4bd3f5ca3cfdbacc66e5dbfe086ae3a3b354a5349e4baae4fcfc0133b1eR140

[kaxil]

@kosteev We might bring it back but yes for now we should add it in Release notes/ migration guide.

We document it as such around using task_ids="my_task_id" when pulling from a single task id:

Check the docs below

• https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/xcoms.html
• https://github.com/apache/airflow/blob/2.10.5/airflow/example_dags/example_xcom.py#L53-L62

[kosteev]

@amoghrajesh @kaxil I think it is different issue, but similar/related.

Here is the issue with using task_ids as a list with one element:

• in Airflow 2 return value will be list with one element
• in Airflow 3 it is just the value itself (not a list)

It has to be here:

airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py

Line 367 in 3ccbcff

return xcoms[0]

As a user I wouldn't expect different type of result depending on the number of elements in the parameter. Is it really intended?