PythonSensor is not considering mode='reschedule', instead marking task UP_FOR_RETRY

[pmrem]

Apache Airflow version

2.3.0 (latest released)

What happened

A PythonSensor that works on versions <2.3.0 in mode reschedule is now marking the task as UP_FOR_RETRY instead.

Log says:

[2022-05-02, 15:48:23 UTC] {python.py:66} INFO - Poking callable: <function test at 0x7fd56286bc10>
[2022-05-02, 15:48:23 UTC] {taskinstance.py:1853} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2022-05-02, 15:48:23 UTC] {local_task_job.py:156} INFO - Task exited with return code 0
[2022-05-02, 15:48:23 UTC] {local_task_job.py:273} INFO - 0 downstream tasks scheduled from follow-on schedule check

But it directly marks it as UP_FOR_RETRY and then follows retry_delay and retries

What you think should happen instead

It should mark the task as UP_FOR_RESCHEDULE and reschedule it according to the poke_interval

How to reproduce

from datetime import datetime, timedelta

from airflow import DAG
from airflow.sensors.python import PythonSensor


def test():
    return False


default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": datetime(2022, 5, 2),
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 1,
    "retry_delay": timedelta(minutes=1),
}

dag = DAG("dag_csdepkrr_development_v001",
          default_args=default_args,
          catchup=False,
          max_active_runs=1,
          schedule_interval=None)

t1 = PythonSensor(task_id="PythonSensor",
                  python_callable=test,
                  poke_interval=30,
                  mode='reschedule',
                  dag=dag)

Operating System

Latest Docker image

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==3.3.0
apache-airflow-providers-celery==2.1.4
apache-airflow-providers-cncf-kubernetes==4.0.1
apache-airflow-providers-docker==2.6.0
apache-airflow-providers-elasticsearch==3.0.3
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-google==6.8.0
apache-airflow-providers-grpc==2.0.4
apache-airflow-providers-hashicorp==2.2.0
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-microsoft-azure==3.8.0
apache-airflow-providers-mysql==2.2.3
apache-airflow-providers-odbc==2.0.4
apache-airflow-providers-oracle==2.2.3
apache-airflow-providers-postgres==4.1.0
apache-airflow-providers-redis==2.0.4
apache-airflow-providers-sendgrid==2.0.4
apache-airflow-providers-sftp==2.5.2
apache-airflow-providers-slack==4.2.3
apache-airflow-providers-sqlite==2.1.3
apache-airflow-providers-ssh==2.4.3

Deployment

Docker-Compose

Deployment details

Latest Docker compose from the documentation

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

[boring-cyborg]

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

[argibbs]

I'm seeing this with airflow.sensors.time_sensor.TimeSensor as well. Renders 2.3.0 unusable for me, unfortunately.

[norm]

I'll take this.

[RobertAbaev]

Hi, got the same behavior with ExternalTaskSensor. Just to make sure it will be noted, thanks!

[bhavaniravi]

Noticed this behavior for a customer for a custom operator. I could recreate it in Astronomer Dev using a simple Python sensor. Contrary to the original ticket I can also see Executor reports task instance finished (success) although the task says its queued. (Info: None) Was the task killed externally?

Dag Code

from airflow.sensors.python import PythonSensor
import time, random
from datetime import datetime
import os
from airflow import DAG

def _partner_b():
    time.sleep(10)
    r = random.random()
    print ("Partner B: {}".format(r))
    return r > 0.8
    
dag = DAG(dag_id=os.path.basename(__file__).replace(".py", ""),
          start_date=datetime(2021, 9, 22),
          default_args={},
          schedule_interval='*/5 * * * *',
          catchup=False)

partner_b = PythonSensor(
    task_id='partner_b',
    poke_interval=60 * 2,
    mode="reschedule",
    python_callable=_partner_b,
    dag=dag,
    weight_rule="absolute"
)

---

Log Trace

[2022-05-09 17:20:11,228] {scheduler_job.py:543} INFO - Sending TaskInstanceKey(dag_id='sensor_example', task_id='partner_b', run_id='scheduled__2022-05-09T17:15:00+00:00', try_number=1, map_index=-1) to executor with priority 1 and queue default
[2022-05-09 17:20:11,228] {base_executor.py:91} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'sensor_example', 'partner_b', 'scheduled__2022-05-09T17:15:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/sensor_example.py']
[2022-05-09 17:20:11,230] {base_executor.py:211} INFO - task TaskInstanceKey(dag_id='sensor_example', task_id='partner_b', run_id='scheduled__2022-05-09T17:15:00+00:00', try_number=1, map_index=-1) is still running
[2022-05-09 17:20:11,230] {scheduler_job.py:596} INFO - Executor reports execution of sensor_example.partner_b run_id=scheduled__2022-05-09T17:15:00+00:00 exited with status success for try_number 1
[2022-05-09 17:20:11,235] {scheduler_job.py:640} INFO - TaskInstance Finished: dag_id=sensor_example, task_id=partner_b, run_id=scheduled__2022-05-09T17:15:00+00:00, map_index=-1, run_start_date=2022-05-09 17:20:00.834723+00:00, run_end_date=2022-05-09 17:20:11.021978+00:00, run_duration=10.187255, state=queued, executor_state=success, try_number=1, max_tries=0, job_id=78, pool=default_pool, queue=default, priority_weight=1, operator=PythonSensor, queued_dttm=2022-05-09 17:20:11.226889+00:00, queued_by_job_id=72, pid=462
[2022-05-09 17:20:11,235] {scheduler_job.py:669} ERROR - Executor reports task instance <TaskInstance: sensor_example.partner_b scheduled__2022-05-09T17:15:00+00:00 [queued]> finished (success) although the task says its queued. (Info: None) Was the task killed externally?
[2022-05-09 17:20:11,237] {taskinstance.py:1890} ERROR - Executor reports task instance <TaskInstance: sensor_example.partner_b scheduled__2022-05-09T17:15:00+00:00 [queued]> finished (success) although the task says its queued. (Info: None) Was the task killed externally?
[2022-05-09 17:20:11,240] {taskinstance.py:1394} INFO - Marking task as FAILED. dag_id=sensor_example, task_id=partner_b, execution_date=20220509T171500, start_date=20220509T172000, end_date=20220509T172011

[tagoad]

Any update on this? a lot of our DAGs are basically unusable because the sensors are getting marked for retry instead of reschedule.

[davidcaron]

I found a fix for this. But I think I would need a bit of help to submit a PR.

davidcaron@21c4d3c

Basically, ReadyToRescheduleDep._get_dep_statuses receives a TaskInstance, but the task property of this task instance is a SerializedBaseOperator object, not the original operator. And this serialized object doesn't have the mode or reschedule attributes, because they are not explicitly included when serializing.

This fix works, but It feels a bit awkward to only include the mode attribute... I'll check tomorrow if I can find more information on what the SerializedBaseOperator is used for.