max_retry_delay should be a timedelta for type hinting

[wyndhblb]

Apache Airflow version:

master

What happened:

https://github.com/apache/airflow/blob/master/airflow/models/baseoperator.py#L356 --> should be timedelta not datetime

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template!

[mik-laj]

Are you willing to submit a PR?

[wyndhblb]

there's another issue i'm working out as well (just found this while investigating that one). even if the max_retry_delay is really a "timedelta" as defined in the task definition

it seems that https://github.com/apache/airflow/blob/master/airflow/models/taskinstance.py#L880 has it as a "float" (and thus the min fails).

[ashb]

Perhaps it should be a float (seconds) then, as this is the only thing that could have possible ever worked?

[iameugenejo]

I'm experiencing the same issue.

scheduler is failing to start up because of this.

{scheduler_job.py:1293} ERROR - Exception when executing SchedulerJob._run_scheduler_loop
Traceback (most recent call last):
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1275, in _execute
    self._run_scheduler_loop()
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1377, in _run_scheduler_loop
    num_queued_tis = self._do_scheduling(session)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1515, in _do_scheduling
    self._schedule_dag_run(dag_run, active_runs_by_dag_id.get(dag_run.dag_id, set()), session)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1677, in _schedule_dag_run
    schedulable_tis, callback_to_run = dag_run.update_state(session=session, execute_callbacks=False)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/models/dagrun.py", line 405, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/models/dagrun.py", line 499, in task_instance_scheduling_decisions
    schedulable_tis, changed_tis = self._get_ready_tis(scheduleable_tasks, finished_tasks, session)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/models/dagrun.py", line 525, in _get_ready_tis
    if st.are_dependencies_met(
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/utils/session.py", line 62, in wrapper
    return func(*args, **kwargs)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 813, in are_dependencies_met
    for dep_status in self.get_failed_dep_statuses(dep_context=dep_context, session=session):
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 834, in get_failed_dep_statuses
    for dep_status in dep.get_dep_statuses(self, session, dep_context):
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/ti_deps/deps/base_ti_dep.py", line 101, in get_dep_statuses
    yield from self._get_dep_statuses(ti, session, dep_context)
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/ti_deps/deps/not_in_retry_period_dep.py", line 47, in _get_dep_statuses
    next_task_retry_date = ti.next_retry_datetime()
  File "/root/airflow/2.0.0/venv/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 881, in next_retry_datetime
    delay = min(self.task.max_retry_delay, delay)
TypeError: '<' not supported between instances of 'datetime.timedelta' and 'float'

[iameugenejo]

patch for /airflow/models/taskinstance.py I'm using to workaround is below:

880c880
<                 delay = min(self.task.max_retry_delay, delay)
---
>                 delay = timedelta(seconds=min(self.task.max_retry_delay, delay.total_seconds()))