apache / apache/airflow

Problem with the sensor file

Open
#58,679 3 comments 0 reactions 0 assignees View on GitHub
area:core kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 10h
Merged PRs (30d)
483

Description

### Apache Airflow version

3.1.3

### If "Other Airflow 2/3 version" selected, which one?

_No response_

### What happened?

I have a base sensor that is designed to reschedule on failed pokes

parameters are as follows

# We will poke every 5 seconds, with exponential backoff, for up to 1 hour
kwargs.setdefault("poke_interval", 5) # start at 5 seconds
kwargs.setdefault("exponential_backoff", True) # enable backoff
# timeout after 1 hour. Time is set in seconds
kwargs.setdefault("timeout", 60 * 60)
# maximum wait between pokes is 1 minute
kwargs.setdefault("max_wait", 60)
# free up worker slots when not poking
kwargs.setdefault("mode", "reschedule")

When I run it via dag.test() i get the following error

2025-11-25T15:34:55.240100Z [error ] Task failed with exception [task] loc=task_runner.py:1002
Traceback (most recent call last):
File "/home/bot/ETL/airflow/venv/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 928, in run
result = _execute_task(context=context, ti=ti, log=log)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/bot/ETL/airflow/venv/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1315, in _execute_task
result = ctx.run(execute, context=context)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/bot/ETL/airflow/venv/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 416, in wrapper
return func(self, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/bot/ETL/airflow/dags/Bridgstone/../../plugins/smb_sensors.py", line 74, in execute
result = super().execute(context)
^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/bot/ETL/airflow/venv/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 416, in wrapper
return func(self, *args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/bot/ETL/airflow/venv/lib/python3.12/site-packages/airflow/sdk/bases/sensor.py", line 235, in execute
if run_duration() > self.timeout:
^^^^^^^^^^^^^^
File "/home/bot/ETL/airflow/venv/lib/python3.12/site-packages/airflow/sdk/bases/sensor.py", line 195, in run_duration
return (timezone.utcnow() - start_date).total_seconds()

### What you think should happen instead?

It seems the sensor file
venv/lib/python3.12/site-packages/airflow/sdk/bases/sensor.py
The first_reschedule_date can be set as a string instead of a datetime
That needs to be fixed

This is the adustment

if self.reschedule:
ti = context["ti"]
first_reschedule_date = ti.get_first_reschedule_date(context)
if isinstance(first_reschedule_date, str):
first_reschedule_date = pendulum.parse(str(first_reschedule_date)) # ensure first_reschedule_date is valid datetime
started_at = start_date = first_reschedule_date or timezone.utcnow()

### How to reproduce

Create dag
Set a base sensor: Set it to reschuale mode

Run the dag via the terminal with dag.test()

### Operating System

Ubuntu 24.04.3 LTS

### Versions of Apache Airflow Providers

apache-airflow-providers-common-compat==1.8.0
apache-airflow-providers-common-io==1.6.4
apache-airflow-providers-common-sql==1.28.2
apache-airflow-providers-fab==2.4.4
apache-airflow-providers-microsoft-mssql==4.3.2
apache-airflow-providers-mysql==6.3.5
apache-airflow-providers-openlineage==2.7.1
apache-airflow-providers-postgres==6.3.0
apache-airflow-providers-smtp==2.3.1
apache-airflow-providers-standard==1.9.1

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

_No response_

### Anything else?

_No response_

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Start with airflow/sdk/bases/sensor.py, especially the reschedule path and run_duration() shown in the traceback, then reproduce the failure with a rescheduling base sensor run through dag.test(). Done means the first-reschedule value is handled without a datetime subtraction error and the sensor can continue rescheduling.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
52/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.