Throw FeatureNotAvailable error when fail_when_dag_is_paused is used in TriggerDagRunOperator with Airflow 3

Author: gopidesupavan

providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py

@@ -65,6 +65,17 @@ def __str__(self) -> str:
         return f"Dag {self.dag_id} is paused"
 
 
+class FeatureNotAvailable(AirflowException):
+    """Raise when a feature is not available."""
+
+    def __init__(self, error_message) -> None:
+        super().__init__(error_message)
+        self.error_message = error_message
+
+    def __str__(self) -> str:
+        return self.error_message
+
+
 class TriggerDagRunLink(BaseOperatorLink):
     """
     Operator link for TriggerDagRunOperator.
@@ -189,6 +200,9 @@ def __init__(
                 f"Expected str, datetime.datetime, or None for parameter 'logical_date'. Got {type(logical_date).__name__}"
             )
 
+        if fail_when_dag_is_paused and AIRFLOW_V_3_0_PLUS:
+            raise FeatureNotAvailable("Setting `fail_when_dag_is_paused` not yet supported for Airflow 3.0")
+
     def execute(self, context: Context):
         if self.logical_date is NOTSET:
             # If no logical_date is provided we will set utcnow()
@@ -218,8 +232,9 @@ def execute(self, context: Context):
         if self.fail_when_dag_is_paused:
             dag_model = DagModel.get_current(self.trigger_dag_id)
             if dag_model.is_paused:
-                if AIRFLOW_V_3_0_PLUS:
-                    raise DagIsPaused(dag_id=self.trigger_dag_id)
+                # TODO: enable this when dag state endpoint available from task sdk
+                # if AIRFLOW_V_3_0_PLUS:
+                #     raise DagIsPaused(dag_id=self.trigger_dag_id)
                 raise AirflowException(f"Dag {self.trigger_dag_id} is paused")
 
         if AIRFLOW_V_3_0_PLUS:

providers/standard/tests/unit/standard/operators/test_trigger_dagrun.py

@@ -31,7 +31,7 @@
 from airflow.models.dagrun import DagRun
 from airflow.models.log import Log
 from airflow.models.taskinstance import TaskInstance
-from airflow.providers.standard.operators.trigger_dagrun import DagIsPaused, TriggerDagRunOperator
+from airflow.providers.standard.operators.trigger_dagrun import FeatureNotAvailable, TriggerDagRunOperator
 from airflow.providers.standard.triggers.external_task import DagStateTrigger
 from airflow.utils.session import create_session
 from airflow.utils.state import DagRunState, TaskInstanceState
@@ -255,6 +255,17 @@ def test_trigger_dag_run_execute_complete_should_fail(self):
                 ),
             )
 
+    def test_trigger_dag_run_with_fail_when_dag_is_paused_should_fail(self):
+        with pytest.raises(
+            FeatureNotAvailable, match="Setting `fail_when_dag_is_paused` not yet supported for Airflow 3.0"
+        ):
+            TriggerDagRunOperator(
+                task_id="test_task",
+                trigger_dag_id=TRIGGERED_DAG_ID,
+                conf={"foo": "bar"},
+                fail_when_dag_is_paused=True,
+            )
+
 
 # TODO: To be removed once the provider drops support for Airflow 2
 @pytest.mark.skipif(AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 2")
@@ -771,9 +782,5 @@ def test_trigger_dagrun_with_fail_when_dag_is_paused(self, dag_maker, session):
                 fail_when_dag_is_paused=True,
             )
         dag_maker.create_dagrun()
-        if AIRFLOW_V_3_0_PLUS:
-            error = DagIsPaused
-        else:
-            error = AirflowException
-        with pytest.raises(error, match=f"^Dag {TRIGGERED_DAG_ID} is paused$"):
+        with pytest.raises(AirflowException, match=f"^Dag {TRIGGERED_DAG_ID} is paused$"):
             task.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)