Tasks marked as "UP_FOR_RESCHEDULE" get stuck in Executor.running and never reschedule

Apache Airflow version

2.3.3

What happened

Upon upgrading from Airflow 2.1.3 to Airflow 2.3.3 we have an issue with our sensors that have mode=‘reschedule’. Using TimeSensor as example:

1. It executes as normal on the first run
2. It detects it is not the correct time yet and marks itself “UP_FOR_RESCEDULE” (usually to rescheduled for 5 minutes in the future)
3. When the time comes to be rescheduled it just gets marked as “QUEUED” and is never actually run again, the error in the log: [2022-08-15 00:01:11,027] {base_executor.py:215} ERROR - could not queue task TaskInstanceKey(dag_id='TestDAG', task_id='testTASK', run_id='scheduled__2022-08-12T04:00:00+00:00', try_number=1, map_index=-1) (still running after 4 attempts)

Looking at the relevant code (https://github.com/apache/airflow/blob/2.3.3/airflow/executors/base_executor.py#L215) it seems that the Task Key was never removed from self.running after it initially rescheduled itself.

What you think should happen instead

Rescheduled tasks should reschedule

How to reproduce

1. Airflow 2.3.3 from Docker
2. Celery 5.2.7 with Redis backend
3. MySQL 8
4. Airflow Timezone set to America/New_York
5. Have a normal (non-async) sensor that has mode reschedule and needs to reschedule itself

Operating System

Fedora 29

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else

The symptoms of this discussion sounds the same, but no one has replied on it yet: https://github.com/apache/airflow/discussions/25651

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