Remove change_sensor_mode_to_reschedule from base executor

airflow-core/docs/core-concepts/executor/index.rst

@@ -278,7 +278,6 @@ The ``BaseExecutor`` class interface contains a set of attributes that Airflow c
 * ``is_single_threaded``: Whether or not the executor is single threaded. This is particularly relevant to what database backends are supported. Single threaded executors can run with any backend, including SQLite.
 * ``is_production``: Whether or not the executor should be used for production purposes. A UI message is displayed to users when they are using a non-production ready executor.
 
-* ``change_sensor_mode_to_reschedule``: Running Airflow sensors in poke mode can block the thread of executors and in some cases Airflow.
 * ``serve_logs``: Whether or not the executor supports serving logs, see :doc:`/administration-and-deployment/logging-monitoring/logging-tasks`.
 
 CLI

airflow-core/src/airflow/executors/base_executor.py

@@ -122,7 +122,6 @@ class BaseExecutor(LoggingMixin):
     is_local: bool = False
     is_production: bool = True
 
-    change_sensor_mode_to_reschedule: bool = False
     serve_logs: bool = False
 
     job_id: None | int | str = None

airflow-core/src/airflow/ti_deps/deps/ready_to_reschedule.py

@@ -50,11 +50,7 @@ def _get_dep_statuses(self, ti, session, dep_context):
         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
+            not is_mapped and not getattr(ti.task, "reschedule", False)
         ):
             yield self._passing_status(reason="Task is not in reschedule mode.")
             return

providers/celery/src/airflow/providers/celery/executors/celery_kubernetes_executor.py

@@ -67,7 +67,6 @@ class CeleryKubernetesExecutor(BaseExecutor):
     is_production: bool = True
 
     serve_logs: bool = False
-    change_sensor_mode_to_reschedule: bool = False
 
     callback_sink: BaseCallbackSink | None = None
 

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/local_kubernetes_executor.py

@@ -58,7 +58,6 @@ class LocalKubernetesExecutor(BaseExecutor):
     is_production: bool = True
 
     serve_logs: bool = True
-    change_sensor_mode_to_reschedule: bool = False
 
     callback_sink: BaseCallbackSink | None = None
 

task-sdk/src/airflow/sdk/bases/sensor.py

@@ -36,13 +36,11 @@
     TaskDeferralError,
     TaskDeferralTimeout,
 )
-from airflow.executors.executor_loader import ExecutorLoader
 from airflow.sdk.bases.operator import BaseOperator
 from airflow.utils import timezone
 
 if TYPE_CHECKING:
     from airflow.sdk.definitions.context import Context
-    from airflow.typing_compat import Self
 
 
 class PokeReturnValue:
@@ -326,18 +324,6 @@ def _get_next_poke_interval(
         self.log.info("new %s interval is %s", self.mode, new_interval)
         return new_interval
 
-    def prepare_for_execution(self) -> Self:
-        task = super().prepare_for_execution()
-
-        # Sensors in `poke` mode can block execution of DAGs when running
-        # with single process executor, thus we change the mode to`reschedule`
-        # to allow parallel task being scheduled and executed
-        executor, _ = ExecutorLoader.import_default_executor_cls()
-        if executor.change_sensor_mode_to_reschedule:
-            self.log.warning("%s changes sensor mode to 'reschedule'.", executor.__name__)
-            task.mode = "reschedule"
-        return task
-
     @property
     def reschedule(self):
         """Define mode rescheduled sensors."""