diff --git a/airflow-core/docs/core-concepts/dags.rst b/airflow-core/docs/core-concepts/dags.rst index ab397f3532c26..63447a0ff5321 100644 --- a/airflow-core/docs/core-concepts/dags.rst +++ b/airflow-core/docs/core-concepts/dags.rst @@ -585,6 +585,48 @@ Dependency relationships can be applied across all tasks in a TaskGroup with the group1() >> task3 +Dependencies between sibling tasks and TaskGroups must remain acyclic when each TaskGroup is treated as a +single unit. This is evaluated using only edges that land on a TaskGroup's root tasks -- tasks with no +upstream task inside that group -- so a cycle can exist at the group level even when the individual task +dependencies form no cycle among themselves: + +.. code-block:: python + + with TaskGroup("group1"): + source1 = EmptyOperator(task_id="source1") + sink1 = EmptyOperator(task_id="sink1") + + with TaskGroup("group2"): + source2 = EmptyOperator(task_id="source2") + sink2 = EmptyOperator(task_id="sink2") + + source1 >> sink2 + source2 >> sink1 + +``sink1`` and ``sink2`` are each a root of their own group (neither has an upstream task inside its own +group), so this places ``group1`` both upstream and downstream of ``group2``. Airflow reports this as a Dag +parsing error even though no individual task-to-task dependency forms a cycle. + +The same "root task" rule can trip up a single group, without a second TaskGroup in sight. A task counts +as a root of its group as soon as it has no upstream task *inside* that group -- even if it has an upstream +task *outside* the group. Routing between two such tasks through an external task closes a cycle on the +group itself: + +.. code-block:: python + + with TaskGroup("group1"): + first = EmptyOperator(task_id="first") + second = EmptyOperator(task_id="second") + + bridge = EmptyOperator(task_id="bridge") + + first >> bridge >> second + +``second`` has no upstream task inside ``group1``, so it is a root of the group even though ``bridge`` sits +outside it. ``group1`` is upstream of ``bridge`` (via ``first``) and downstream of ``bridge`` (via +``second``'s root edge), which Airflow rejects the same way as the sibling-group example above -- even +though this reads like an ordinary Dag with a single TaskGroup. + TaskGroup also supports ``default_args`` like Dag, it will overwrite the ``default_args`` in Dag level: .. code-block:: python diff --git a/airflow-core/newsfragments/73087.significant.rst b/airflow-core/newsfragments/73087.significant.rst new file mode 100644 index 0000000000000..a42e7cd319555 --- /dev/null +++ b/airflow-core/newsfragments/73087.significant.rst @@ -0,0 +1,19 @@ +Cyclic TaskGroup dependencies now fail Dag parsing instead of failing later in Grid or Graph + +Dag parsing now rejects TaskGroups whose dependencies form a cycle when each group is treated as a single +unit -- for example, a task in ``group1`` upstream of ``group2``'s root task while a task in ``group2`` is +upstream of ``group1``'s root task. Task-level dependencies can be acyclic while this group-level +projection still cycles, and downstream consumers such as Grid and Graph need an unambiguous group +ordering. The same root-task rule can also close a cycle on a single group: a task with no upstream task +inside its own group counts as a root of that group even when an external task provides its only upstream, +so routing between two such tasks through an outside task closes a cycle on the group itself. + +Before this change, a Dag with cyclic TaskGroup dependencies could parse and run its tasks normally. #69933 +fixed the group-level topological sort used by Grid and Graph to actually detect group-to-group and +cross-group edges -- it previously ignored them and rendered an arbitrary, incorrect order instead of +failing. That made a pre-existing cycle in a Dag's TaskGroups surface as an unhandled error from the Grid +and Graph API endpoints rather than a rendering quirk. This change moves that failure to Dag parsing time +instead, where it raises an import error naming the offending TaskGroup and the specific task/TaskGroup ids +involved in the cycle, rather than a 500 response from the UI. + +Dags that previously parsed with cyclic TaskGroup dependencies must be restructured to remove the cycle. diff --git a/airflow-core/tests/unit/dag_processing/test_dagbag.py b/airflow-core/tests/unit/dag_processing/test_dagbag.py index cb110ef5c7fda..0856f6e304fa5 100644 --- a/airflow-core/tests/unit/dag_processing/test_dagbag.py +++ b/airflow-core/tests/unit/dag_processing/test_dagbag.py @@ -1025,6 +1025,37 @@ def basic_cycle(): self.validate_dags(test_dag, found_dags, dagbag, should_be_found=False) assert file_path in dagbag.import_errors + def test_skip_task_group_dependency_cycle_dags(self, tmp_path): + def task_group_dependency_cycle(): + import datetime + + from airflow.providers.standard.operators.empty import EmptyOperator + from airflow.sdk import DAG, TaskGroup + + with DAG( + "task_group_dependency_cycle", + schedule=None, + start_date=datetime.datetime(2016, 1, 1), + ) as dag: + with TaskGroup("left"): + left_source = EmptyOperator(task_id="left_source") + left_sink = EmptyOperator(task_id="left_sink") + with TaskGroup("right"): + right_source = EmptyOperator(task_id="right_source") + right_sink = EmptyOperator(task_id="right_sink") + + left_source >> right_sink + right_source >> left_sink + + return dag + + test_dag = task_group_dependency_cycle() + + dagbag, found_dags, file_path = self.process_dag(task_group_dependency_cycle, tmp_path) + + self.validate_dags(test_dag, found_dags, dagbag, should_be_found=False) + assert "TaskGroup dependency cycle detected" in dagbag.import_errors[file_path] + def test_process_file_with_none(self, tmp_path): """ test that process_file can handle Nones diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py b/task-sdk/src/airflow/sdk/definitions/dag.py index 633feccb0483a..f02841ce4d449 100644 --- a/task-sdk/src/airflow/sdk/definitions/dag.py +++ b/task-sdk/src/airflow/sdk/definitions/dag.py @@ -1138,7 +1138,7 @@ def add_result(self, xcom_arg: X) -> X: def check_cycle(self) -> None: """ - Check to see if there are any cycles in the Dag. + Check to see if there are any task or TaskGroup dependency cycles in the Dag. :raises AirflowDagCycleException: If cycle is found in the Dag. """ @@ -1177,6 +1177,22 @@ def _check_adjacent_tasks(task_id, current_task): else: path_stack.append(child_to_check) + task_group_dict = self.task_group.get_task_group_dict() + for task_group in task_group_dict.values(): + try: + task_group.topological_sort(group_dict=task_group_dict) + except AirflowDagCycleException as cycle_exc: + group_id = task_group.group_id or "" + nodes_detail = ( + f" Nodes involved: {', '.join(cycle_exc.cyclic_node_ids)}" + if cycle_exc.cyclic_node_ids + else "" + ) + raise AirflowDagCycleException( + f"TaskGroup dependency cycle detected in Dag: {self.dag_id}. " + f"Faulty TaskGroup: {group_id}.{nodes_detail}" + ) from None + def cli(self): """Exposes a CLI specific to this Dag.""" self.check_cycle() diff --git a/task-sdk/src/airflow/sdk/definitions/taskgroup.py b/task-sdk/src/airflow/sdk/definitions/taskgroup.py index 89bc37d7127bd..df441d2bd1684 100644 --- a/task-sdk/src/airflow/sdk/definitions/taskgroup.py +++ b/task-sdk/src/airflow/sdk/definitions/taskgroup.py @@ -657,7 +657,9 @@ def _sweep_projection(self, nodes: list[DAGNode], projected: list[tuple[int, ... emitted[i] = 1 order_append(nodes[i]) if len(next_pending) == len(pending): - raise AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc = AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc.cyclic_node_ids = tuple(sorted(nodes[i].node_id for i in pending)) + raise exc pending = next_pending return order @@ -697,7 +699,9 @@ def _sort_via_pass_numbering( queue.append(s) if processed != n: - raise AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc = AirflowDagCycleException(f"A cyclic dependency occurred in dag: {self.dag_id}") + exc.cyclic_node_ids = tuple(sorted(nodes[i].node_id for i in range(n) if in_degree[i] != 0)) + raise exc sorted_indices = sorted(range(n), key=lambda i: (pass_of[i], i)) return [nodes[i] for i in sorted_indices] diff --git a/task-sdk/src/airflow/sdk/exceptions.py b/task-sdk/src/airflow/sdk/exceptions.py index 6f43d5421ecf2..2f9f3b8d17712 100644 --- a/task-sdk/src/airflow/sdk/exceptions.py +++ b/task-sdk/src/airflow/sdk/exceptions.py @@ -71,6 +71,9 @@ class AirflowSecretsBackendAccessDenied(PermissionError): class AirflowDagCycleException(AirflowException): """Raise when there is a cycle in Dag definition.""" + #: node ids (tasks and/or TaskGroups) identified as part of the cycle, when known. + cyclic_node_ids: tuple[str, ...] = () + class AirflowRuntimeError(Exception): """Generic Airflow error raised by runtime functions.""" diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py b/task-sdk/tests/task_sdk/definitions/test_dag.py index 07b8c8186c8e5..05c44af420108 100644 --- a/task-sdk/tests/task_sdk/definitions/test_dag.py +++ b/task-sdk/tests/task_sdk/definitions/test_dag.py @@ -997,6 +997,64 @@ def test_cycle_task_group_with_edge_labels(self): assert not dag.check_cycle() + @pytest.mark.parametrize( + ("nested", "faulty_group", "faulty_nodes"), + [ + pytest.param(False, "", "left, right", id="root"), + pytest.param(True, "parent", "parent.left, parent.right", id="nested"), + ], + ) + def test_cycle_between_sibling_task_groups(self, nested, faulty_group, faulty_nodes): + dag = DAG("dag", schedule=None, start_date=DEFAULT_DATE, default_args={"owner": "owner1"}) + + def add_groups(): + with TaskGroup("left"): + left_source = DoNothingOperator(task_id="left_source") + left_sink = DoNothingOperator(task_id="left_sink") + with TaskGroup("right"): + right_source = DoNothingOperator(task_id="right_source") + right_sink = DoNothingOperator(task_id="right_sink") + + left_source >> right_sink + right_source >> left_sink + + with dag: + if nested: + with TaskGroup("parent"): + add_groups() + else: + add_groups() + + with pytest.raises( + AirflowDagCycleException, + match=rf"TaskGroup dependency cycle detected in Dag: dag\. Faulty TaskGroup: {faulty_group}\. " + rf"Nodes involved: {faulty_nodes}", + ): + dag.check_cycle() + + def test_cycle_between_group_root_bridged_by_external_task(self): + """A task outside a TaskGroup can create a cycle by bridging two of the group's own tasks. + + ``second`` has no upstream task inside ``group1``, so it counts as a root of the group even + though its only upstream (``bridge``) sits outside it. + """ + dag = DAG("dag", schedule=None, start_date=DEFAULT_DATE, default_args={"owner": "owner1"}) + + with dag: + with TaskGroup("group1"): + first = DoNothingOperator(task_id="first") + second = DoNothingOperator(task_id="second") + bridge = DoNothingOperator(task_id="bridge") + + first >> bridge >> second + + with pytest.raises( + AirflowDagCycleException, + match=r"TaskGroup dependency cycle detected in Dag: dag\. Faulty TaskGroup: \. " + r"Nodes involved: bridge, group1", + ): + dag.check_cycle() + class TestDagGetItem: def test_getitem_returns_task(self):