TimeDeltaSensorAsync is failing when logical_date is null

[atul-astronomer]

Apache Airflow version

3.0.0b1

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

No response

What happened?

TimeDeltaSensorAsync is failing when logical_date is null

Task logs:

[2025-03-05T13:48:34.522473Z] ERROR - Task failed with exception logger="task" error_detail=[{"exc_type":"KeyError","exc_value":"'data_interval_end'","syntax_error":null,"is_cause":false,"frames":[{"filename":"/opt/airflow/task_sdk/src/airflow/sdk/execution_time/task_runner.py","lineno":605,"name":"run"},{"filename":"/opt/airflow/task_sdk/src/airflow/sdk/execution_time/task_runner.py","lineno":726,"name":"_execute_task"},{"filename":"/opt/airflow/airflow/models/baseoperator.py","lineno":168,"name":"wrapper"},{"filename":"/opt/airflow/providers/standard/src/airflow/providers/standard/sensors/time_delta.py","lineno":95,"name":"execute"}]}]

What you think should happen instead?

TimeDeltaSensorAsync should pass when logical_date is null

How to reproduce

Trigger the below DAG with logical_date as null:

from __future__ import annotations

import datetime

import pendulum

from airflow import DAG
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.sensors.time_delta import TimeDeltaSensorAsync

with DAG(
    dag_id="example_time_delta_sensor_async",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:
    wait = TimeDeltaSensorAsync(task_id="wait", delta=datetime.timedelta(seconds=30))

    finish = EmptyOperator(task_id="finish")
    wait >> finish

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