[airflow] branch main updated: Fixes to how DebugExecutor handles sensors (#28528)

[onikolas]

This is an automated email from the ASF dual-hosted git repository.

onikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new a35ec9549d Fixes to how DebugExecutor handles sensors (#28528)
a35ec9549d is described below

commit a35ec9549dc0b4b311f126f3022309ec5b33fa62
Author: Raphaël Vandon <114772123+vandonr-...@users.noreply.github.com>
AuthorDate: Tue Jan 24 21:40:10 2023 -0800

    Fixes to how DebugExecutor handles sensors (#28528)
    
    * move fix to ready_to_reschedule
    * replace check on debug exec with new property
    
    Co-authored-by: Tzu-ping Chung <uranu...@gmail.com>
---
 airflow/executors/debug_executor.py         |  6 ++++++
 airflow/ti_deps/deps/ready_to_reschedule.py | 15 +++++++++++----
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/airflow/executors/debug_executor.py 
b/airflow/executors/debug_executor.py
index 18ba0f8798..4355bd1dcf 100644
--- a/airflow/executors/debug_executor.py
+++ b/airflow/executors/debug_executor.py
@@ -25,6 +25,7 @@ DebugExecutor.
 from __future__ import annotations
 
 import threading
+import time
 from typing import Any
 
 from airflow.configuration import conf
@@ -120,6 +121,11 @@ class DebugExecutor(BaseExecutor):
 
         :param open_slots: Number of open slots
         """
+        if not self.queued_tasks:
+            # wait a bit if there are no tasks ready to be executed to avoid 
spinning too fast in the void
+            time.sleep(0.5)
+            return
+
         sorted_queue = sorted(
             self.queued_tasks.items(),
             key=lambda x: x[1][1],
diff --git a/airflow/ti_deps/deps/ready_to_reschedule.py 
b/airflow/ti_deps/deps/ready_to_reschedule.py
index 467f08536e..66aa5c5613 100644
--- a/airflow/ti_deps/deps/ready_to_reschedule.py
+++ b/airflow/ti_deps/deps/ready_to_reschedule.py
@@ -17,6 +17,7 @@
 # under the License.
 from __future__ import annotations
 
+from airflow.executors.executor_loader import ExecutorLoader
 from airflow.models.taskreschedule import TaskReschedule
 from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
 from airflow.utils import timezone
@@ -44,10 +45,16 @@ class ReadyToRescheduleDep(BaseTIDep):
         from airflow.models.mappedoperator import MappedOperator
 
         is_mapped = isinstance(ti.task, MappedOperator)
-        if not is_mapped and not getattr(ti.task, "reschedule", False):
-            # Mapped sensors don't have the reschedule property (it can only
-            # be calculated after unmapping), so we don't check them here.
-            # They are handled below by checking TaskReschedule instead.
+        executor, _ = ExecutorLoader.import_default_executor_cls()
+        if (
+            # Mapped sensors don't have the reschedule property (it can only 
be calculated after unmapping),
+            # so we don't check them here. They are handled below by checking 
TaskReschedule instead.
+            not is_mapped
+            and not getattr(ti.task, "reschedule", False)
+            # Executors can force running in reschedule mode,
+            # in which case we ignore the value of the task property.
+            and not executor.change_sensor_mode_to_reschedule
+        ):
             yield self._passing_status(reason="Task is not in reschedule 
mode.")
             return