include_prior_dates=True not finding XCom values within the same DAG in 3.0.3

[ldacey]

Apache Airflow version

3.0.3

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

No response

What happened?

I have many DAGs which incrementally download data by passing a max_value XCom (modified timestamp for files, updated_at timestamp etc).

When I upgraded to Airflow 3.0.3 I noticed that this stopped working and downgrading to 3.0.2 worked again. I don't have many details other than a few logs:

Run 1:

INFO - Pushing 'extracted_files' XCom with 10 files
INFO - Pushing 'max_value' XCom: 2025-07-20 20:00:27+00:00 - dag_id: campaign_extract_raw, task_id: extract_source_to_raw,
run_id: scheduled__2025-07-10T04:02:00+00:00

Run 2 (XCom is pulling the same DAG ID and task ID)

Attempting XCom pull - dag_id: campaign_extract_raw, task_id: extract_source_to_raw
INFO - XCom pull result (current run) for max_value: None
INFO - XCom pull result (with prior dates) for max_value
INFO - All XCom values for task: None

---

Another example where I logged anything I could think of to try to narrow down the issue. The DAG was correctly pulling from the same DAG ID and task ID. None was returned and the DAG fell back to the start_date (so attempted to redownload the entire table...). This DAG is returning the max value from a taskflow task, so the key is return_value instead:

INFO - Cloning repository: bundle_name="gitlab-legacy": version=null:
INFO - Filling up the DagBag from /opt/airflow/dag_bundles/gitlab-legacy/tracking_repo/src/dags/
INFO - get_previous_max_value called with task_id=extract_source_to_raw, key=return_value:
INFO - DAG params: {'since': None, 'until': None}:
INFO - Current DAG ID: example_extract_raw:
INFO - Using DAG ID: example_extract_raw, task ID: extract_source_to_raw:
INFO - Attempting XCom pull for dag_id=example_extract_raw, task_id=extract_source_to_raw, key=return_value, include_prior_dates=True:
INFO - XCom pull result: None (type: <class 'NoneType'>):
INFO - XCom pull returned None or empty value:
INFO - Using DAG start date as fallback: 2017-12-31T18:30:00+00:00:

The next run After downgrading to Airflow 3.0.2 correctly found the XCom:

INFO - Using DAG ID: example_extract_raw, task ID: extract_source_to_raw
INFO - Attempting XCom pull for dag_id=example_extract_raw, task_id=extract_source_to_raw, key=return_value, include_prior_dates=True
INFO - XCom pull result: 2025-07-20T00:00:00 (type: <class 'str'>)
INFO - Found previous max value: 2025-07-20T00:00:00

What you think should happen instead?

include_prior_dates should find the XCom from within the same DAG. I checked the xcom DB table and they are definitely there.

How to reproduce

Upgrade to 3.0.3 and run a DAG with include_prior_dates retrieving a key from within the same DAG.

Operating System

Debian bookworm

Versions of Apache Airflow Providers

• apache-airflow-providers-cncf-kubernetes==10.6.1
• apache-airflow-providers-common-compat==1.7.2
• apache-airflow-providers-common-io==1.6.1
• apache-airflow-providers-common-sql==1.27.3
• apache-airflow-providers-fab==2.3.0
• apache-airflow-providers-ftp==3.13.1
• apache-airflow-providers-google==16.1.0
• apache-airflow-providers-http==5.3.2
• apache-airflow-providers-microsoft-azure==12.5.0
• apache-airflow-providers-odbc==4.10.1
• apache-airflow-providers-openlineage==2.5.0
• apache-airflow-providers-postgres==6.2.1
• apache-airflow-providers-salesforce==5.11.1
• apache-airflow-providers-sftp==5.3.2
• apache-airflow-providers-smtp==2.1.1
• apache-airflow-providers-ssh==4.1.1
• apache-airflow-providers-standard==1.4.1

Deployment

Official Apache Airflow Helm Chart

Deployment details

I used the 1.18.0 Airflow Helm chart and tested locally with Docker compose.

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

[ashb]

Are you able to share any kind of dag? (even if privately, this would help to get to the bottom of this)

[ashb]

Are you using mapped tasks on either end?

[ldacey]

Are you using mapped tasks on either end?

No - no mapped tasks at all.

My DAG files are registered using config files so they don't show much.

Here is an older one where the lower_bound is the pulling the timestamp through XCom from the previous run directly in a DAG template file (less abstraction compared to the examples above):

    @dag(**_get_base_dag_args(config))
    def extract_dag() -> None:
        """Pipeline to extract data from database incrementally."""
        from common.taskflow.extract.database import extract_records
        from common.taskflow.triggers.transform import trigger_transformation_dag

        @task(
            task_id="notify_extraction_start",
            task_display_name="Notify Webex",
        )
        def notify_extraction_start() -> None:
            """Notifies the Webex Teams space that the pipeline has started."""

        start = notify_extraction_start()

        extracted_data = extract_records.override(
            task_id="extract_records",
            task_display_name="Query Database Records",
            executor_config=config.get_resources(),
        )(
            config=config,
            lower_bound="{{ params.lower_bound or ti.xcom_pull(task_ids='extract_records', key='max_value', include_prior_dates=True) or dag.start_date }}",
        )

        start >> extracted_data

        trigger_transformation_dag.override(
            task_id="trigger_transformation_dag",
            task_display_name="Trigger Transform",
            outlets=[config.get_destination_dataset()().airflow_asset],
        )(
            config=config,
            files=extracted_data,
            message_id="{{ ti.xcom_pull(task_ids='notify_extraction_start') }}",
        )

    return extract_dag()

And the XCom view in the UI has the correct XCom and the max_value key which should have been pulled.

Airflow 3.0.2 (lower bound is correct from previous run max_value)

{"timestamp":"2025-07-16T16:07:24.464260","level":"info","event":"Final query parameters: {'lower_bound': '2025-07-15 18:10:18', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

{"timestamp":"2025-07-17T16:07:32.238572","level":"info","event":"Final query parameters: {'lower_bound': '2025-07-16 18:08:48', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

Airflow 3.0.3 (lower bound is the DAG start). I did not immediately notice the issue because the DAG still ran successfully:

{"timestamp":"2025-07-18T16:07:22.463658","level":"info","event":"Final query parameters: {'lower_bound': '2022-12-31 18:30:00', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

{"timestamp":"2025-07-19T16:07:48.413275","level":"info","event":"Final query parameters: {'lower_bound': '2022-12-31 18:30:00', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

{"timestamp":"2025-07-20T16:17:58.726883","level":"info","event":"Final query parameters: {'lower_bound': '2022-12-31 18:30:00', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

Airflow 3.0.2 (downgraded DB and reinstalled 3.0.2):

{"timestamp":"2025-07-21T03:47:44.358213","level":"info","event":"Final query parameters: {'lower_bound': '2025-07-19 18:06:54', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

{"timestamp":"2025-07-21T16:07:25.134960","level":"info","event":"Final query parameters: {'lower_bound': '2025-07-20 18:17:46', 'max_records': 8500000, 'incremental_column': 'processeddatetime'}","logger":"common.models.extract.database"}

[ashb]

@ldacey Do you have any xcom backend configured?

[ldacey]

I didn't originally when I first encountered the issue but I added it as part of the troubleshooting and it is still in place.

  - name: AIRFLOW__CORE__XCOM_BACKEND
    value: "airflow.providers.common.io.xcom.backend.XComObjectStorageBackend"
  - name: AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH
    value: "gs://conn-id@airflow_bucket/xcom"
  - name: AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD
    value: "1000"

Added it on 07/20 but that is when I downgraded to 3.0.2 as well. I do not think this was the fix because I got XComs to work locally with Docker compose before I updated the helm chart (no xcom_backend in docker).

Date:   Sun Jul 20 07:31:06 2025 +0800

    fix(airflow): downgrade airflow to 3.0.2 and enable xcom_backend

Timestamp of when I upgraded to 3.0.3 lines up with the logs above where the DAGs began pulling since dag_start instead of the XCom value as well:

defaultAirflowTag: 3.0.3
defaultAirflowTag: 3.0.2

Date:   Fri Jul 18 02:45:14 2025 +0800

    ci(airflow): bump to 3.0.3

[amoskaliov]

I confirm this bug appeared in 3.0.3. There wasn't in 3.0.2.

Here is the basic example:

from datetime import datetime, timedelta

from airflow.sdk import dag, task

@dag(
    schedule=timedelta(seconds=60),
    start_date=datetime(2025, 7, 22),
    catchup=False,
    max_active_runs=1
)
def xcom_test():

    @task
    def xcom_task(ti):
        data_xcom = ti.xcom_pull(key=f"test_key", include_prior_dates=True)

        if data_xcom:
            print(f'XCom pulled. Value: {data_xcom}')
        else:
            print(f'XCom not found')

        ti.xcom_push(key=f"test_key", value=' - xcom value - ')

    xcom_task()

xcom_test()

During the 2nd run you'll get XCom not found, although it was pushed during the 1st run.

[dlira-sf]

I was testing an upgrade to Airflow 3.0.3 from Airflow 3.0.2, and I encountered the same issue. Only extra info I can add is I also saw these in XCom pulls happening across different DAGs. I am sure it is the same bug, but just leaving this here to give more info about where I see the error happening.