This is an automated email from the ASF dual-hosted git repository.
dheerajturaga 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 4ca1f6693bb Deprecate cyclic TaskGroup dependencies (#73746)
4ca1f6693bb is described below
commit 4ca1f6693bb380ae1f148a6187001adbcadbfa22
Author: Dheeraj Turaga <[email protected]>
AuthorDate: Sun Oct 11 08:00:35 2026 -0500
Deprecate cyclic TaskGroup dependencies (#73746)
---
airflow-core/docs/core-concepts/dags.rst | 72 +++++++
airflow-core/newsfragments/73746.significant.rst | 22 ++
.../core_api/openapi/v2-rest-api-generated.yaml | 1 +
.../src/airflow/dag_processing/collection.py | 1 +
airflow-core/src/airflow/dag_processing/dagbag.py | 31 ++-
.../src/airflow/dag_processing/processor.py | 3 +
airflow-core/src/airflow/models/dagwarning.py | 1 +
.../airflow/ui/openapi-gen/requests/schemas.gen.ts | 2 +-
.../airflow/ui/openapi-gen/requests/types.gen.ts | 2 +-
.../core_api/routes/public/test_dag_warning.py | 6 +-
.../tests/unit/dag_processing/test_collection.py | 24 +++
.../tests/unit/dag_processing/test_dagbag.py | 58 +++++-
.../tests/unit/dag_processing/test_processor.py | 35 ++++
.../src/airflowctl/api/datamodels/generated.py | 1 +
task-sdk/src/airflow/sdk/definitions/dag.py | 45 ++++
task-sdk/src/airflow/sdk/definitions/taskgroup.py | 45 +++-
task-sdk/src/airflow/sdk/exceptions.py | 4 +
task-sdk/tests/task_sdk/definitions/test_dag.py | 228 ++++++++++++++++++++-
18 files changed, 563 insertions(+), 18 deletions(-)
diff --git a/airflow-core/docs/core-concepts/dags.rst
b/airflow-core/docs/core-concepts/dags.rst
index 724619aae9f..9946e17bb72 100644
--- a/airflow-core/docs/core-concepts/dags.rst
+++ b/airflow-core/docs/core-concepts/dags.rst
@@ -624,6 +624,78 @@ If you want to see a more advanced use of TaskGroup, you
can look at the ``examp
When using the ``@task_group`` decorator, the decorated-function's
docstring will be used as the TaskGroups tooltip in the UI except when a
``tooltip`` value is explicitly supplied.
+.. _concepts:taskgroup-cycles:
+
+Cyclic TaskGroup dependencies
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+.. deprecated:: 3.4.0
+ Dags with cyclic TaskGroup dependencies are planned to fail Dag parsing
from Airflow 3.5.
+
+When each TaskGroup is treated as a single unit, a TaskGroup and its siblings
must not depend on each other
+in a cycle. A dependency into or out of any task in a group counts as a
dependency of the whole group. This
+can make a group both upstream and downstream of a sibling, although no
task-level dependency forms a cycle:
+
+.. code-block:: python
+
+ with TaskGroup("group1"):
+ a1 = EmptyOperator(task_id="a1")
+ a2 = EmptyOperator(task_id="a2")
+
+ with TaskGroup("group2"):
+ b1 = EmptyOperator(task_id="b1")
+ b2 = EmptyOperator(task_id="b2")
+
+ a1 >> b1 # group2 depends on group1
+ b2 >> a2 # group1 depends on group2
+
+A path that leaves a TaskGroup and comes back into it also forms a cycle, even
when the tasks inside the
+group are ordered directly as well:
+
+.. code-block:: python
+
+ with TaskGroup("group"):
+ a = EmptyOperator(task_id="a")
+ b = EmptyOperator(task_id="b")
+ a >> b
+
+ bridge = EmptyOperator(task_id="bridge")
+
+ a >> bridge >> b # group -> bridge -> group
+
+This includes a setup and teardown pair inside a TaskGroup with the work they
wrap outside it. This is
+intended: a TaskGroup that holds a setup and its teardown must also hold the
tasks between them.
+
+.. code-block:: python
+
+ with TaskGroup("cluster"):
+ create = EmptyOperator(task_id="create")
+ delete = EmptyOperator(task_id="delete")
+
+ work = EmptyOperator(task_id="work")
+
+ create >> work >> delete.as_teardown(setups=create) # cluster -> work ->
cluster
+
+These Dags still parse and run, but features that act on a TaskGroup as a
whole need an unambiguous order
+between groups. Parsing them emits a ``TaskGroupCycleDeprecationWarning`` and
a Dag warning in the UI that
+name the TaskGroups and tasks involved. The warning lists at most five members
of each cycle and at most ten
+cycles, with a count of the ones left out.
+
+To remove the cycle, move tasks between TaskGroups, or out of them, so that
each group depends on the others
+in one direction only. In the second example, move ``bridge`` into ``group``,
or move ``b`` out of it. In the
+setup and teardown example, move ``work`` into ``cluster``.
+
+Only Dags authored with the Python Task SDK are checked. Dags from the Go,
Java and TypeScript SDKs are not
+checked yet.
+
+To catch these Dags in CI, turn the warning into an error. The check runs when
a Dag is added to a
+``DagBag``, so this works for a pytest test that loads your Dags into a
``DagBag`` and asserts there are no
+import errors, but not for one that only imports a Dag file:
+
+.. code-block:: bash
+
+ pytest -W error::airflow.sdk.exceptions.TaskGroupCycleDeprecationWarning
tests/test_dags.py
+
.. _concepts:edge-labels:
Edge Labels
diff --git a/airflow-core/newsfragments/73746.significant.rst
b/airflow-core/newsfragments/73746.significant.rst
new file mode 100644
index 00000000000..6d976dfc2b5
--- /dev/null
+++ b/airflow-core/newsfragments/73746.significant.rst
@@ -0,0 +1,22 @@
+Deprecate cyclic TaskGroup dependencies
+
+A Dag whose TaskGroups depend on each other in a cycle, when each TaskGroup is
treated as a single unit,
+now issues a ``TaskGroupCycleDeprecationWarning`` at parse time and shows a
Dag warning in the UI that
+names the TaskGroups and tasks involved. These Dags still parse and run,
because their task dependencies
+are acyclic, but Airflow 3.5 is planned to reject them at parse time.
+
+A dependency into or out of any task in a TaskGroup counts as a dependency of
the whole group, so a path
+that leaves a TaskGroup and comes back into it is a cycle. For example, ``a >>
bridge >> b``, with ``a``
+and ``b`` in the same TaskGroup and ``bridge`` outside it, makes the group
depend on ``bridge`` and
+``bridge`` depend on the group, even if ``a >> b`` is also set.
+
+**Migration:**
+
+- Move tasks between TaskGroups, or out of them, so that each group depends on
the others in one direction
+ only. In the example above, move ``bridge`` into the group, or move ``b``
out of it.
+- To catch these Dags in CI, run your Dag tests with
+ ``-W error::airflow.sdk.exceptions.TaskGroupCycleDeprecationWarning``. The
check runs when a Dag is
+ added to a ``DagBag``, so only tests that load your Dags into a ``DagBag``
and assert there are no import
+ errors catch them; a test that only imports a Dag file does not.
+
+See "Cyclic TaskGroup dependencies" in the Dags documentation.
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index 9ca0cf9db1e..28b9241bc06 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -16250,6 +16250,7 @@ components:
- duplicate dag id
- non-existent pool
- runtime varying value
+ - task group cycle
title: DagWarningType
description: 'Enum for DAG warning types.
diff --git a/airflow-core/src/airflow/dag_processing/collection.py
b/airflow-core/src/airflow/dag_processing/collection.py
index 3e20722c6c2..cd7228d210f 100644
--- a/airflow-core/src/airflow/dag_processing/collection.py
+++ b/airflow-core/src/airflow/dag_processing/collection.py
@@ -587,6 +587,7 @@ def update_dag_parsing_results_in_db(
DagWarningType.DUPLICATE_DAG_ID,
DagWarningType.NONEXISTENT_POOL,
DagWarningType.RUNTIME_VARYING_VALUE,
+ DagWarningType.TASK_GROUP_CYCLE,
),
files_parsed: set[tuple[str, str]] | None = None,
dag_source_codes: dict[str, DagSourceCode] | None = None,
diff --git a/airflow-core/src/airflow/dag_processing/dagbag.py
b/airflow-core/src/airflow/dag_processing/dagbag.py
index 3b68817758d..dc80433676c 100644
--- a/airflow-core/src/airflow/dag_processing/dagbag.py
+++ b/airflow-core/src/airflow/dag_processing/dagbag.py
@@ -46,6 +46,7 @@ from airflow.exceptions import (
from airflow.executors.executor_loader import ExecutorLoader
from airflow.listeners.listener import get_listener_manager
from airflow.models.pool import Pool
+from airflow.sdk.exceptions import AirflowDagCycleException,
TaskGroupCycleDeprecationWarning
from airflow.sdk.importers import DagImportError, get_importer_registry
from airflow.serialization.definitions.dag import SerializedDAG
from airflow.serialization.definitions.notset import NOTSET, ArgNotSet,
is_arg_set
@@ -213,6 +214,7 @@ class DagBag(LoggingMixin):
self.import_errors: dict[str, str] = {}
self.captured_warnings: dict[str, tuple[str, ...]] = {}
self._import_warnings: dict[str, list[DagImportWarning]] = {}
+ self.task_group_cycle_warnings: dict[str, str] = {}
# The source code of each definition that produced Dags, keyed by its
fileloc
self.dag_source_codes: dict[str, DagSourceCode] = {}
# Only used by SchedulerJob to compare the dag_hash to identify change
in DAGs
@@ -446,7 +448,11 @@ class DagBag(LoggingMixin):
"""Get the set of DagWarnings for the bagged dags."""
from airflow.models.dagwarning import DagWarning, DagWarningType
- dag_warnings: set[DagWarning] = set()
+ dag_warnings: set[DagWarning] = {
+ DagWarning(dag_id, DagWarningType.TASK_GROUP_CYCLE, message)
+ for dag_id, message in self.task_group_cycle_warnings.items()
+ if dag_id in self.dags
+ }
for dag in self.dags.values():
for import_warning in self._import_warnings.get(dag.fileloc, ()):
# Only importer-namespaced types (``yaml:deprecated_field``)
are Dag warnings;
@@ -491,7 +497,21 @@ class DagBag(LoggingMixin):
:raises: AirflowDagCycleException if a cycle is detected.
:raises: AirflowDagDuplicatedIdException if this dag already exists in
the bag.
"""
- dag.check_cycle()
+ task_group_cycle_warning = None
+ with warnings.catch_warnings(record=True) as captured_warnings:
+ # DeprecationWarning is ignored by default outside __main__, which
would hide it here too.
+ warnings.simplefilter("always", TaskGroupCycleDeprecationWarning)
+ dag.check_cycle()
+ for captured in captured_warnings:
+ if issubclass(captured.category, TaskGroupCycleDeprecationWarning):
+ task_group_cycle_warning = str(captured.message)
+ warnings.warn_explicit(
+ message=captured.message,
+ category=captured.category,
+ filename=captured.filename,
+ lineno=captured.lineno,
+ source=captured.source,
+ )
dag.resolve_template_files()
dag.last_loaded = timezone.utcnow()
@@ -515,6 +535,11 @@ class DagBag(LoggingMixin):
self.log.exception(e)
raise AirflowClusterPolicyError(e)
self._add_to_bag(dag)
+ # Only once bagged: a rejected duplicate must not touch the warning of
the Dag it lost to.
+ if task_group_cycle_warning is None:
+ self.task_group_cycle_warnings.pop(dag.dag_id, None)
+ else:
+ self.task_group_cycle_warnings[dag.dag_id] =
task_group_cycle_warning
def _add_to_bag(self, dag: BaggedDAG) -> None:
"""
@@ -522,8 +547,6 @@ class DagBag(LoggingMixin):
:raises: AirflowDagDuplicatedIdException if this dag already exists in
the bag.
"""
- from airflow.sdk.exceptions import AirflowDagCycleException
-
try:
prev_dag = self.dags.get(dag.dag_id)
if prev_dag and prev_dag.fileloc != dag.fileloc:
diff --git a/airflow-core/src/airflow/dag_processing/processor.py
b/airflow-core/src/airflow/dag_processing/processor.py
index c3feb3f4959..e6f3f422a0a 100644
--- a/airflow-core/src/airflow/dag_processing/processor.py
+++ b/airflow-core/src/airflow/dag_processing/processor.py
@@ -254,6 +254,8 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
serialized_dags, serialization_import_errors = _serialize_dags(bag, log)
bag.import_errors.update(serialization_import_errors)
+ # Dag warnings reference the dag table, so only Dags that serialized can
carry one.
+ serialized_dag_ids = {dag.dag_id for dag in serialized_dags}
result = DagFileParsingResult(
fileloc=msg.file,
serialized_dags=serialized_dags,
@@ -263,6 +265,7 @@ def _parse_file(msg: DagFileParseRequest, log:
FilteringBoundLogger) -> DagFileP
*(
{"dag_id": w.dag_id, "warning_type": w.warning_type,
"message": w.message}
for w in bag.dag_warnings
+ if w.dag_id in serialized_dag_ids
),
],
parsed_definitions=bag.parsed_definitions,
diff --git a/airflow-core/src/airflow/models/dagwarning.py
b/airflow-core/src/airflow/models/dagwarning.py
index 6305361d2aa..4d1eeac7db5 100644
--- a/airflow-core/src/airflow/models/dagwarning.py
+++ b/airflow-core/src/airflow/models/dagwarning.py
@@ -108,6 +108,7 @@ class DagWarningType(str, Enum):
DUPLICATE_DAG_ID = "duplicate dag id"
NONEXISTENT_POOL = "non-existent pool"
RUNTIME_VARYING_VALUE = "runtime varying value"
+ TASK_GROUP_CYCLE = "task group cycle"
ImporterWarningType = Annotated[
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
index f3651041f84..68ccfe0903b 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts
@@ -5618,7 +5618,7 @@ export const $DagVersionResponse = {
export const $DagWarningType = {
type: 'string',
- enum: ['asset conflict', 'duplicate dag id', 'non-existent pool', 'runtime
varying value'],
+ enum: ['asset conflict', 'duplicate dag id', 'non-existent pool', 'runtime
varying value', 'task group cycle'],
title: 'DagWarningType',
description: `Enum for DAG warning types.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index c27c63231e5..bc19ad00938 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -1555,7 +1555,7 @@ export type DagVersionResponse = {
* This is the set of allowable values for the ``warning_type`` field
* in the DagWarning model.
*/
-export type DagWarningType = 'asset conflict' | 'duplicate dag id' |
'non-existent pool' | 'runtime varying value';
+export type DagWarningType = 'asset conflict' | 'duplicate dag id' |
'non-existent pool' | 'runtime varying value' | 'task group cycle';
/**
* How much of a component's work has a live instance covering it.
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
index 7a15de834c9..e96311a666c 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_warning.py
@@ -115,9 +115,9 @@ class TestGetDagWarnings:
response = test_client.get("/dagWarnings", params={"warning_type":
"invalid"})
response_json = response.json()
assert response.status_code == 422
- assert (
- response_json["detail"][0]["msg"]
- == "Input should be 'asset conflict', 'duplicate dag id',
'non-existent pool' or 'runtime varying value'"
+ assert response_json["detail"][0]["msg"] == (
+ "Input should be 'asset conflict', 'duplicate dag id',
'non-existent pool', "
+ "'runtime varying value' or 'task group cycle'"
)
def test_get_dag_warnings_non_matching_type(self, test_client):
diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py
b/airflow-core/tests/unit/dag_processing/test_collection.py
index 8a7fe712bb5..6e9a61c8756 100644
--- a/airflow-core/tests/unit/dag_processing/test_collection.py
+++ b/airflow-core/tests/unit/dag_processing/test_collection.py
@@ -994,6 +994,30 @@ class TestUpdateDagParsingResults:
dag_code = DagCode.get_latest_dagcode("yaml_dag", session=session)
assert (dag_code.source_code, dag_code.language) == ("dag_id:
yaml_dag\n", "yaml")
+ def test_task_group_cycle_warning_is_removed_when_dag_is_fixed(self,
testing_dag_bundle, session):
+ dag = DAG(dag_id="fixed_dag")
+ session.add(DagModel(dag_id="fixed_dag", bundle_name="testing",
is_stale=False))
+ session.add(
+ DagWarning(
+ dag_id="fixed_dag",
+ warning_type=DagWarningType.TASK_GROUP_CYCLE,
+ message="Dag 'fixed_dag': group and bridge depend on each
other in a cycle.",
+ )
+ )
+ session.flush()
+
+ update_dag_parsing_results_in_db(
+ bundle_name="testing",
+ bundle_version=None,
+ dags=[LazyDeserializedDAG.from_dag(dag)],
+ import_errors={},
+ parse_duration=None,
+ warnings=set(),
+ session=session,
+ )
+
+ assert session.scalar(select(DagWarning).where(DagWarning.dag_id ==
"fixed_dag")) is None
+
def test_parse_time_written_to_db_on_sync(self, testing_dag_bundle,
session):
"""Test that the parse time is correctly written to the DB after
parsing"""
diff --git a/airflow-core/tests/unit/dag_processing/test_dagbag.py
b/airflow-core/tests/unit/dag_processing/test_dagbag.py
index dc02e9ea085..73a718580f0 100644
--- a/airflow-core/tests/unit/dag_processing/test_dagbag.py
+++ b/airflow-core/tests/unit/dag_processing/test_dagbag.py
@@ -44,15 +44,15 @@ from airflow.dag_processing.dagbag import (
)
from airflow.dag_processing.lang_sdk_processor import
LangSDKDagFileProcessorProcess
from airflow.dag_processing.processor import DagFileParsingResult
-from airflow.exceptions import UnknownExecutorException
+from airflow.exceptions import AirflowDagDuplicatedIdException,
UnknownExecutorException
from airflow.executors.executor_loader import ExecutorLoader
from airflow.models.dag import DagModel
from airflow.models.dagwarning import DagWarning, DagWarningType
from airflow.models.errors import ParseImportError
from airflow.models.pool import Pool
from airflow.models.serialized_dag import SerializedDagModel
-from airflow.sdk import DAG, BaseOperator
-from airflow.sdk.exceptions import AirflowConfigException
+from airflow.sdk import DAG, BaseOperator, TaskGroup
+from airflow.sdk.exceptions import AirflowConfigException,
TaskGroupCycleDeprecationWarning
from airflow.sdk.importers import (
AbstractDagImporter,
DagImportResult,
@@ -1402,6 +1402,58 @@ with airflow.DAG(
os.fspath(dag_file): (f"{dag_file}:1: test:deprecated_field:
Deprecated field",)
}
+ @staticmethod
+ def _make_task_group_cycle_dag(*, cyclic: bool) -> DAG:
+ with DAG(dag_id="test") as dag:
+ with TaskGroup("group") as group:
+ a = BaseOperator(task_id="a")
+ b = BaseOperator(task_id="b")
+ a >> BaseOperator(task_id="bridge", task_group=None if cyclic else
group) >> b
+ return dag
+
+ def test_dag_warnings_task_group_cycle(self):
+ dagbag = DagBag(dag_folder="", collect_dags=False)
+
+ # Python ignores DeprecationWarning outside __main__, as in the Dag
processor.
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore")
+ dagbag.bag_dag(self._make_task_group_cycle_dag(cyclic=True))
+
+ (warning,) = dagbag.dag_warnings
+ assert (warning.dag_id, warning.warning_type) == ("test",
DagWarningType.TASK_GROUP_CYCLE)
+ assert "group and bridge depend on each other in a cycle." in
warning.message
+
+ with pytest.warns(TaskGroupCycleDeprecationWarning):
+ dagbag.bag_dag(self._make_task_group_cycle_dag(cyclic=True))
+
+ dagbag.bag_dag(self._make_task_group_cycle_dag(cyclic=False))
+ assert dagbag.dag_warnings == set()
+
+ @pytest.mark.parametrize(
+ ("bagged_cyclic", "duplicate_cyclic"),
+ [
+ pytest.param(False, True, id="cyclic-duplicate-of-acyclic"),
+ pytest.param(True, False, id="acyclic-duplicate-of-cyclic"),
+ ],
+ )
+ def test_dag_warnings_task_group_cycle_ignores_rejected_duplicate(self,
bagged_cyclic, duplicate_cyclic):
+ dagbag = DagBag(dag_folder="", collect_dags=False)
+ bagged = self._make_task_group_cycle_dag(cyclic=bagged_cyclic)
+ bagged.fileloc = "/dags/bagged.py"
+ duplicate = self._make_task_group_cycle_dag(cyclic=duplicate_cyclic)
+ duplicate.fileloc = "/dags/duplicate.py"
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore")
+ dagbag.bag_dag(bagged)
+ with pytest.raises(AirflowDagDuplicatedIdException):
+ dagbag.bag_dag(duplicate)
+
+ assert dagbag.dags["test"] is bagged
+ assert {w.warning_type for w in dagbag.dag_warnings} == (
+ {DagWarningType.TASK_GROUP_CYCLE} if bagged_cyclic else set()
+ )
+
def test_sigsegv_handling(self, tmp_path, caplog):
"""
Test that a SIGSEGV in a DAG file is handled gracefully and does not
crash the process.
diff --git a/airflow-core/tests/unit/dag_processing/test_processor.py
b/airflow-core/tests/unit/dag_processing/test_processor.py
index f5abf5f8b52..629cb6b1c49 100644
--- a/airflow-core/tests/unit/dag_processing/test_processor.py
+++ b/airflow-core/tests/unit/dag_processing/test_processor.py
@@ -794,6 +794,41 @@ def
test_parse_file_reports_definitions_source_codes_and_dag_warnings(mock_dag_w
]
+TASK_GROUP_CYCLE_DAGS = """
+from airflow.sdk import DAG, BaseOperator, TaskGroup
+
+for dag_id in ("cyclic", "acyclic", "unserializable"):
+ with DAG(dag_id, schedule=None):
+ with TaskGroup("group") as group:
+ a = BaseOperator(task_id="a")
+ b = BaseOperator(task_id="b")
+ a >> BaseOperator(task_id="bridge", task_group=group if dag_id ==
"acyclic" else None) >> b
+"""
+
+
+def test_parse_file_task_group_cycle_warning(tmp_path):
+ dag_file = tmp_path / "task_group_cycle_dags.py"
+ dag_file.write_text(TASK_GROUP_CYCLE_DAGS)
+ to_dict = DagSerialization.to_dict
+
+ def fail_to_serialize_unserializable(dag):
+ if dag.dag_id == "unserializable":
+ raise ValueError("serialization failed")
+ return to_dict(dag)
+
+ with patch.object(
+ DagSerialization, "to_dict", autospec=True,
side_effect=fail_to_serialize_unserializable
+ ):
+ result = _parse_file(
+ DagFileParseRequest(file=str(dag_file), bundle_path=tmp_path,
bundle_name="testing"),
+ log=structlog.get_logger(),
+ )
+
+ assert [(warning["dag_id"], warning["warning_type"]) for warning in
result.warnings] == [
+ ("cyclic", "task group cycle")
+ ]
+
+
def test_callback_processing_does_not_update_timestamps():
"""Callback processing should not update last_finish_time to prevent stale
DAG detection."""
stat = process_parse_results(
diff --git a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
index edef5aed5a0..337a83c76cc 100644
--- a/airflow-ctl/src/airflowctl/api/datamodels/generated.py
+++ b/airflow-ctl/src/airflowctl/api/datamodels/generated.py
@@ -911,6 +911,7 @@ class DagWarningType(str, Enum):
DUPLICATE_DAG_ID = "duplicate dag id"
NON_EXISTENT_POOL = "non-existent pool"
RUNTIME_VARYING_VALUE = "runtime varying value"
+ TASK_GROUP_CYCLE = "task group cycle"
class DetailedHealthStatus(str, Enum):
diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py
b/task-sdk/src/airflow/sdk/definitions/dag.py
index 14ddc937dda..0594ac80658 100644
--- a/task-sdk/src/airflow/sdk/definitions/dag.py
+++ b/task-sdk/src/airflow/sdk/definitions/dag.py
@@ -59,6 +59,7 @@ from airflow.sdk.exceptions import (
NodeNotFound,
ParamValidationError,
RemovedInAirflow4Warning,
+ TaskGroupCycleDeprecationWarning,
TaskNotFound,
)
@@ -87,6 +88,11 @@ log = logging.getLogger(__name__)
TAG_MAX_LEN = 100
+# The TaskGroup cycle warning becomes a DagWarning.message, a Text column (64
KB on MySQL), so its
+# length must not grow with the size or the number of the cycles.
+_MAX_LISTED_TASK_GROUP_CYCLES = 10
+_MAX_LISTED_TASK_GROUP_CYCLE_MEMBERS = 5
+
__all__ = [
"DAG",
"dag",
@@ -305,6 +311,13 @@ def _default_task_group(instance: DAG) -> TaskGroup:
return TaskGroup.create_root(dag=instance)
+def _format_cycle_members(node_ids: list[str]) -> str:
+ if len(node_ids) > _MAX_LISTED_TASK_GROUP_CYCLE_MEMBERS:
+ unlisted = len(node_ids) - _MAX_LISTED_TASK_GROUP_CYCLE_MEMBERS
+ return f"{', '.join(node_ids[:_MAX_LISTED_TASK_GROUP_CYCLE_MEMBERS])}
and {unlisted} more"
+ return f"{', '.join(node_ids[:-1])} and {node_ids[-1]}"
+
+
def _is_valid_dag_result(value: Any) -> TypeIs[PlainXComArg]:
from airflow.sdk.bases.xcom import BaseXCom
from airflow.sdk.definitions.xcom_arg import PlainXComArg
@@ -1149,6 +1162,9 @@ class DAG:
"""
Check to see if there are any cycles in the Dag.
+ Also warns with ``TaskGroupCycleDeprecationWarning`` when TaskGroups
depend on each
+ other in a cycle, which is deprecated.
+
:raises AirflowDagCycleException: If cycle is found in the Dag.
"""
task_dict = self.task_dict
@@ -1158,6 +1174,35 @@ class DAG:
f"Cycle detected in Dag: {self.dag_id}. Faulty task:
{faulty_task_id}"
)
+ self._warn_task_group_cycles()
+
+ def _warn_task_group_cycles(self) -> None:
+ task_group_dict = self.task_group.get_task_group_dict()
+ # With only the root group, the projection is the task graph, which is
acyclic here.
+ if len(task_group_dict) == 1:
+ return
+ cycles = [
+ cycle
+ for task_group in task_group_dict.values()
+ for cycle in
task_group._find_dependency_cycles(group_dict=task_group_dict)
+ ]
+ if not cycles:
+ return
+ listed = "; ".join(_format_cycle_members(cycle) for cycle in
cycles[:_MAX_LISTED_TASK_GROUP_CYCLES])
+ unlisted = len(cycles) - _MAX_LISTED_TASK_GROUP_CYCLES
+ unlisted_note = (
+ f" ({unlisted} more cycle{'s' if unlisted > 1 else ''} not
listed)" if unlisted > 0 else ""
+ )
+ # Airflow 3.5 raises AirflowDagCycleException here instead; tracked at
+ # https://github.com/apache/airflow/issues/73678
+ warnings.warn(
+ f"Dag '{self.dag_id}': {listed} depend on each other in a
cycle{unlisted_note}. Cyclic TaskGroup "
+ "dependencies are deprecated and will fail Dag parsing in Airflow
3.5. See "
+ '"Cyclic TaskGroup dependencies" in the docs.',
+ TaskGroupCycleDeprecationWarning,
+ stacklevel=3,
+ )
+
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 21417e75b01..90689d1309d 100644
--- a/task-sdk/src/airflow/sdk/definitions/taskgroup.py
+++ b/task-sdk/src/airflow/sdk/definitions/taskgroup.py
@@ -23,7 +23,7 @@ import copy
import re
import weakref
from collections import deque
-from collections.abc import Generator, Iterator, Sequence
+from collections.abc import Collection, Generator, Iterator, Sequence
from typing import TYPE_CHECKING, Any
import attrs
@@ -595,7 +595,15 @@ class TaskGroup(TaskGroupMixin, DAGNode):
id_to_idx: dict[str, int],
group_dict: dict[str, TaskGroup],
) -> tuple[int, ...]:
- upstream_ids = child._topological_upstream_ids
+ return self._project_upstream_ids(child_idx,
child._topological_upstream_ids, id_to_idx, group_dict)
+
+ def _project_upstream_ids(
+ self,
+ child_idx: int,
+ upstream_ids: Collection[str],
+ id_to_idx: dict[str, int],
+ group_dict: dict[str, TaskGroup],
+ ) -> tuple[int, ...]:
if not upstream_ids:
return ()
sib_deps: set[int] = set()
@@ -669,6 +677,39 @@ class TaskGroup(TaskGroupMixin, DAGNode):
return self._sort_cyclic_projection(nodes, projected)
return [nodes[i] for i in sorted_indices]
+ def _find_dependency_cycles(self, *, group_dict: dict[str, TaskGroup]) ->
list[list[str]]:
+ """
+ Find children that depend on each other in a cycle when each child
TaskGroup is one unit.
+
+ An edge into any task of a child group counts as an edge into the
group, so a path that
+ leaves a group and comes back into it is a cycle even though the
task-level graph is
+ acyclic. This is stricter than the ordering ``topological_sort``
needs, which only looks
+ at edges into a group's roots.
+
+ :return: one list of child node ids per cycle, each in insertion order
+ """
+ nodes = list(self.children.values())
+ id_to_idx = {nid: i for i, nid in enumerate(self.children)}
+ projected = [
+ self._project_upstream_ids(i, self._get_unit_upstream_ids(child),
id_to_idx, group_dict)
+ for i, child in enumerate(nodes)
+ ]
+ members: dict[int, list[str]] = {}
+ for i, component in
enumerate(self._find_projection_components(projected)):
+ members.setdefault(component, []).append(nodes[i].node_id)
+ return [node_ids for node_ids in members.values() if len(node_ids) > 1]
+
+ @staticmethod
+ def _get_unit_upstream_ids(child: DAGNode) -> Collection[str]:
+ if not isinstance(child, TaskGroup):
+ return child._topological_upstream_ids
+ # Not _topological_upstream_ids or iter(child):
MappedTaskGroup.__iter__ raises for a
+ # trigger_rule="always" child, which would turn an acyclic Dag into an
import error.
+ upstream_ids = set(child.upstream_task_ids)
+ upstream_ids.update(group_id for group_id in child.upstream_group_ids
if group_id is not None)
+ upstream_ids.update(edge_id for task in child.iter_tasks() for edge_id
in task.upstream_task_ids)
+ return upstream_ids
+
def iter_mapped_task_groups(self) -> Iterator[MappedTaskGroup]:
"""
Return mapped task groups in the hierarchy.
diff --git a/task-sdk/src/airflow/sdk/exceptions.py
b/task-sdk/src/airflow/sdk/exceptions.py
index 6f43d5421ec..78a7c491c99 100644
--- a/task-sdk/src/airflow/sdk/exceptions.py
+++ b/task-sdk/src/airflow/sdk/exceptions.py
@@ -427,3 +427,7 @@ class RemovedInAirflow4Warning(DeprecationWarning):
deprecated_since: str | None = None
"Indicates the airflow version that started raising this deprecation
warning"
+
+
+class TaskGroupCycleDeprecationWarning(DeprecationWarning):
+ """Issued for Dags whose TaskGroups depend on each other in a cycle;
Airflow 3.5 will reject them."""
diff --git a/task-sdk/tests/task_sdk/definitions/test_dag.py
b/task-sdk/tests/task_sdk/definitions/test_dag.py
index 66dc9545d50..be9975f1b93 100644
--- a/task-sdk/tests/task_sdk/definitions/test_dag.py
+++ b/task-sdk/tests/task_sdk/definitions/test_dag.py
@@ -33,6 +33,7 @@ from airflow.sdk import (
Param,
PartitionedAtRuntime,
TaskGroup,
+ TriggerRule,
dag as dag_decorator,
task,
task_group,
@@ -41,7 +42,12 @@ from airflow.sdk.bases.operator import BaseOperator
from airflow.sdk.bases.timetable import BaseTimetable
from airflow.sdk.definitions._internal.loop import create_loop
from airflow.sdk.definitions.param import DagParam, ParamsDict
-from airflow.sdk.exceptions import AirflowDagCycleException,
DuplicateTaskIdFound, RemovedInAirflow4Warning
+from airflow.sdk.exceptions import (
+ AirflowDagCycleException,
+ DuplicateTaskIdFound,
+ RemovedInAirflow4Warning,
+ TaskGroupCycleDeprecationWarning,
+)
from airflow.utils.types import DagRunType
DEFAULT_DATE = datetime(2016, 1, 1, tzinfo=UTC)
@@ -83,14 +89,15 @@ class TestDag:
assert dag.topological_sort() == ()
def test_dag_topological_sort_task_group_cycle(self):
- """A TaskGroup that depends on a sibling in a cycle, although
check_cycle passes, still sorts."""
+ """A TaskGroup that depends on a sibling in a cycle, which check_cycle
only warns about, still sorts."""
with DAG("dag", schedule=None, start_date=DEFAULT_DATE) as dag:
bridge = BaseOperator(task_id="bridge")
with TaskGroup("group"):
a = BaseOperator(task_id="a")
b = BaseOperator(task_id="b")
a >> bridge >> b
- dag.check_cycle()
+ with pytest.warns(TaskGroupCycleDeprecationWarning):
+ dag.check_cycle()
assert [task.task_id for task in dag.topological_sort()] == ["bridge",
"group.a", "group.b"]
@@ -473,7 +480,6 @@ class TestDag:
def test_dag_add_task_checks_trigger_rule(self):
# A non fail stop dag should allow any trigger rule
- from airflow.sdk import TriggerRule
from airflow.sdk.exceptions import FailFastDagInvalidTriggerRule
class CustomOperator(BaseOperator):
@@ -988,6 +994,134 @@ class DoNothingOperator(BaseOperator):
pass
+def _add_sibling_groups_cycle():
+ with TaskGroup("group1"):
+ a1 = DoNothingOperator(task_id="a1")
+ a2 = DoNothingOperator(task_id="a2")
+ with TaskGroup("group2"):
+ b1 = DoNothingOperator(task_id="b1")
+ b2 = DoNothingOperator(task_id="b2")
+ a1 >> b1
+ b2 >> a2
+ return [a1, b2], [a2, b1]
+
+
+def _make_sibling_groups_cycle_dag():
+ with DAG("dag", schedule=None) as dag:
+ firsts, lasts = _add_sibling_groups_cycle()
+ DoNothingOperator(task_id="start") >> firsts
+ lasts >> DoNothingOperator(task_id="end")
+ return dag
+
+
+def _make_group_bridged_by_outside_task_dag():
+ with DAG("dag", schedule=None) as dag:
+ with TaskGroup("group"):
+ a = DoNothingOperator(task_id="a")
+ b = DoNothingOperator(task_id="b")
+ a >> DoNothingOperator(task_id="bridge") >> b
+ return dag
+
+
+def _make_group_reentered_after_in_group_upstream_dag():
+ with DAG("dag", schedule=None) as dag:
+ with TaskGroup("group"):
+ a = DoNothingOperator(task_id="a")
+ b = DoNothingOperator(task_id="b")
+ a >> b
+ a >> DoNothingOperator(task_id="bridge") >> b
+ return dag
+
+
+def _make_sibling_groups_cycle_through_non_roots_dag():
+ with DAG("dag", schedule=None) as dag:
+ with TaskGroup("group1"):
+ a1 = DoNothingOperator(task_id="a1")
+ a2 = DoNothingOperator(task_id="a2")
+ a1 >> a2
+ with TaskGroup("group2"):
+ b1 = DoNothingOperator(task_id="b1")
+ b2 = DoNothingOperator(task_id="b2")
+ b1 >> b2
+ a1 >> b2
+ b1 >> a2
+ return dag
+
+
+def _make_three_group_ring_dag():
+ with DAG("dag", schedule=None) as dag:
+ tasks = {}
+ for group_id in ("g0", "g1", "g2"):
+ with TaskGroup(group_id):
+ tasks[group_id] = (DoNothingOperator(task_id="first"),
DoNothingOperator(task_id="second"))
+ tasks["g1"][0] >> tasks["g0"][1]
+ tasks["g2"][0] >> tasks["g1"][1]
+ tasks["g0"][0] >> tasks["g2"][1]
+ return dag
+
+
+def _make_two_cycles_dag():
+ with DAG("dag", schedule=None) as dag:
+ _add_sibling_groups_cycle()
+ with TaskGroup("outer"):
+ _add_sibling_groups_cycle()
+ return dag
+
+
+def _make_setup_teardown_dag(*, work_in_group: bool):
+ with DAG("dag", schedule=None) as dag:
+ with TaskGroup("cluster") as cluster:
+ create = DoNothingOperator(task_id="create")
+ delete = DoNothingOperator(task_id="delete")
+ work = DoNothingOperator(task_id="work", task_group=cluster if
work_in_group else None)
+ create >> work >> delete.as_teardown(setups=create)
+ return dag
+
+
+def _make_setup_teardown_around_many_tasks_dag():
+ with DAG("dag", schedule=None) as dag:
+ with TaskGroup("cluster"):
+ create = DoNothingOperator(task_id="create")
+ delete = DoNothingOperator(task_id="delete")
+ work = [DoNothingOperator(task_id=f"work_{i}") for i in range(100)]
+ create >> work >> delete.as_teardown(setups=create)
+ return dag
+
+
+def _make_many_bridged_groups_dag(count: int):
+ with DAG("dag", schedule=None) as dag:
+ for i in range(count):
+ with TaskGroup(f"group_{i}"):
+ a = DoNothingOperator(task_id="a")
+ b = DoNothingOperator(task_id="b")
+ a >> DoNothingOperator(task_id=f"bridge_{i}") >> b
+ return dag
+
+
+def _make_group_with_bridge_inside_dag():
+ with DAG("dag", schedule=None) as dag:
+ with TaskGroup("group"):
+ a = DoNothingOperator(task_id="a")
+ b = DoNothingOperator(task_id="b")
+ a >> DoNothingOperator(task_id="bridge") >> b
+ return dag
+
+
+def _make_mapped_group_with_always_task_dag():
+ with DAG("dag", schedule=None) as dag:
+
+ @task(trigger_rule=TriggerRule.ALWAYS)
+ def work(value):
+ return value
+
+ @task_group
+ def group(value):
+ work(value)
+
+ group.expand(value=[1, 2, 3])
+ return dag
+
+
class TestCycleTester:
def test_cycle_empty(self):
# test empty
@@ -1129,6 +1263,92 @@ class TestCycleTester:
assert not dag.check_cycle()
+ @pytest.mark.parametrize(
+ ("make_dag", "expected_cycles"),
+ [
+ pytest.param(_make_sibling_groups_cycle_dag, "group1 and group2",
id="sibling-groups"),
+ pytest.param(_make_group_bridged_by_outside_task_dag, "group and
bridge", id="bridged-group"),
+ pytest.param(
+ _make_group_reentered_after_in_group_upstream_dag,
+ "group and bridge",
+ id="group-reentered-after-in-group-upstream",
+ ),
+ pytest.param(
+ _make_sibling_groups_cycle_through_non_roots_dag,
+ "group1 and group2",
+ id="sibling-groups-through-non-roots",
+ ),
+ pytest.param(_make_three_group_ring_dag, "g0, g1 and g2",
id="three-group-ring"),
+ pytest.param(
+ _make_two_cycles_dag,
+ "group1 and group2; outer.group1 and outer.group2",
+ id="root-and-nested-cycles",
+ ),
+ pytest.param(
+ lambda: _make_setup_teardown_dag(work_in_group=False),
+ "cluster and work",
+ id="setup-teardown-around-outside-task",
+ ),
+ ],
+ )
+ def test_task_group_cycle_warns(self, make_dag, expected_cycles):
+ dag = make_dag()
+
+ with pytest.warns(TaskGroupCycleDeprecationWarning) as record:
+ dag.check_cycle()
+
+ assert len(record) == 1
+ assert str(record[0].message).startswith(
+ f"Dag 'dag': {expected_cycles} depend on each other in a cycle. "
+ )
+
+ @pytest.mark.parametrize(
+ ("make_dag", "expected_start"),
+ [
+ pytest.param(
+ _make_setup_teardown_around_many_tasks_dag,
+ "Dag 'dag': cluster, work_0, work_1, work_2, work_3 and 96
more depend on each other in a "
+ "cycle. ",
+ id="cycle-members",
+ ),
+ pytest.param(
+ lambda: _make_many_bridged_groups_dag(11),
+ f"Dag 'dag': {'; '.join(f'group_{i} and bridge_{i}' for i in
range(10))} depend on each "
+ "other in a cycle (1 more cycle not listed). ",
+ id="one-unlisted-cycle",
+ ),
+ pytest.param(
+ lambda: _make_many_bridged_groups_dag(12),
+ f"Dag 'dag': {'; '.join(f'group_{i} and bridge_{i}' for i in
range(10))} depend on each "
+ "other in a cycle (2 more cycles not listed). ",
+ id="unlisted-cycles",
+ ),
+ ],
+ )
+ def test_task_group_cycle_warning_caps_listed_ids(self, make_dag,
expected_start):
+ dag = make_dag()
+
+ with pytest.warns(TaskGroupCycleDeprecationWarning) as record:
+ dag.check_cycle()
+
+ assert len(record) == 1
+ assert str(record[0].message).startswith(expected_start)
+
+ @pytest.mark.parametrize(
+ "make_dag",
+ [
+ pytest.param(_make_group_with_bridge_inside_dag,
id="bridge-inside-group"),
+ pytest.param(lambda: _make_setup_teardown_dag(work_in_group=True),
id="setup-teardown-in-group"),
+ pytest.param(_make_mapped_group_with_always_task_dag,
id="mapped-group-with-always-task"),
+ ],
+ )
+ def test_acyclic_task_groups_do_not_warn(self, make_dag):
+ dag = make_dag()
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("error", TaskGroupCycleDeprecationWarning)
+ dag.check_cycle()
+
class TestDagGetItem:
def test_getitem_returns_task(self):
