Fix ExternalTaskSensor to use timeout parameter in deferrable mode (#…

providers/standard/src/airflow/providers/standard/sensors/external_task.py

@@ -432,10 +432,15 @@ def execute(self, context: Context) -> None:
         if not self.deferrable:
             super().execute(context)
         else:
+            # Determine the timeout to use: prefer timeout parameter, fallback to execution_timeout
+            timeout_value = self.timeout
+            if not timeout_value and self.execution_timeout:
+                timeout_value = self.execution_timeout.total_seconds()
+
             dttm_filter = self._get_dttm_filter(context)
             if AIRFLOW_V_3_0_PLUS:
                 self.defer(
-                    timeout=self.execution_timeout,
+                    timeout=datetime.timedelta(seconds=timeout_value) if timeout_value else None,
                     trigger=WorkflowTrigger(
                         external_dag_id=self.external_dag_id,
                         external_task_group_id=self.external_task_group_id,
@@ -453,7 +458,7 @@ def execute(self, context: Context) -> None:
                 )
             else:
                 self.defer(
-                    timeout=self.execution_timeout,
+                    timeout=datetime.timedelta(seconds=timeout_value) if timeout_value else None,
                     trigger=WorkflowTrigger(
                         external_dag_id=self.external_dag_id,
                         external_task_group_id=self.external_task_group_id,

providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py

@@ -1049,6 +1049,77 @@ def test_external_task_sensor_deferrable(self, dag_maker):
         assert exc.value.trigger.external_task_ids == ["test_task"]
         assert exc.value.trigger.execution_dates == [DEFAULT_DATE]
 
+    @pytest.mark.execution_timeout(10)
+    def test_external_task_sensor_deferrable_timeout_only(self, dag_maker):
+        """Test that deferrable mode uses timeout parameter when only timeout is set."""
+        context = {"execution_date": DEFAULT_DATE}
+        with dag_maker() as dag:
+            op = ExternalTaskSensor(
+                task_id="test_external_task_sensor_check",
+                external_dag_id="test_dag_parent",
+                external_task_id="test_task",
+                deferrable=True,
+                timeout=60,
+            )
+            dr = dag.create_dagrun(
+                run_id="test_run",
+                run_type=DagRunType.MANUAL,
+                state=None,
+            )
+            context.update(dag_run=dr, logical_date=DEFAULT_DATE)
+
+        with pytest.raises(TaskDeferred) as exc:
+            op.execute(context=context)
+        assert exc.value.timeout == timedelta(seconds=60)
+
+    @pytest.mark.execution_timeout(10)
+    def test_external_task_sensor_deferrable_execution_timeout_only(self, dag_maker):
+        """Test that deferrable mode falls back to execution_timeout when timeout is not set."""
+        context = {"execution_date": DEFAULT_DATE}
+        with dag_maker() as dag:
+            op = ExternalTaskSensor(
+                task_id="test_external_task_sensor_check",
+                external_dag_id="test_dag_parent",
+                external_task_id="test_task",
+                deferrable=True,
+                timeout=0,  # Explicitly set to 0 to indicate not using timeout
+                execution_timeout=timedelta(seconds=120),
+            )
+            dr = dag.create_dagrun(
+                run_id="test_run",
+                run_type=DagRunType.MANUAL,
+                state=None,
+            )
+            context.update(dag_run=dr, logical_date=DEFAULT_DATE)
+
+        with pytest.raises(TaskDeferred) as exc:
+            op.execute(context=context)
+        assert exc.value.timeout == timedelta(seconds=120)
+
+    @pytest.mark.execution_timeout(10)
+    def test_external_task_sensor_deferrable_timeout_priority(self, dag_maker):
+        """Test that deferrable mode prioritizes timeout over execution_timeout when both are set."""
+        context = {"execution_date": DEFAULT_DATE}
+        with dag_maker() as dag:
+            op = ExternalTaskSensor(
+                task_id="test_external_task_sensor_check",
+                external_dag_id="test_dag_parent",
+                external_task_id="test_task",
+                deferrable=True,
+                timeout=90,
+                execution_timeout=timedelta(seconds=120),
+            )
+            dr = dag.create_dagrun(
+                run_id="test_run",
+                run_type=DagRunType.MANUAL,
+                state=None,
+            )
+            context.update(dag_run=dr, logical_date=DEFAULT_DATE)
+
+        with pytest.raises(TaskDeferred) as exc:
+            op.execute(context=context)
+        assert exc.value.timeout == timedelta(seconds=90)
+
     def test_get_logical_date(self):
         """For AF 2, we check for execution_date in context."""
         context = {"execution_date": DEFAULT_DATE}