EbrahimKaram opened a new issue, #58679:
URL: https://github.com/apache/airflow/issues/58679
### 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)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
