You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by sh...@apache.org on 2026/07/23 18:49:19 UTC
(airflow) branch main updated: Stop skipping none_failed_min_one_success tasks in mapped task groups (#69377)
This is an automated email from the ASF dual-hosted git repository.
shahar1 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 dd029a89953 Stop skipping none_failed_min_one_success tasks in mapped task groups (#69377)
dd029a89953 is described below
commit dd029a899537c61709226af4d4b0f514fb1c025c
Author: Shahar Epstein <60...@users.noreply.github.com>
AuthorDate: Thu Jul 23 21:49:11 2026 +0300
Stop skipping none_failed_min_one_success tasks in mapped task groups (#69377)
---
airflow-core/newsfragments/69377.bugfix.rst | 1 +
.../src/airflow/ti_deps/deps/trigger_rule_dep.py | 28 +++++---
.../tests/unit/models/test_mappedoperator.py | 79 ++++++++++++++++++++++
3 files changed, 98 insertions(+), 10 deletions(-)
diff --git a/airflow-core/newsfragments/69377.bugfix.rst b/airflow-core/newsfragments/69377.bugfix.rst
new file mode 100644
index 00000000000..bfbb5f560f6
--- /dev/null
+++ b/airflow-core/newsfragments/69377.bugfix.rst
@@ -0,0 +1 @@
+A task with the ``none_failed_min_one_success`` trigger rule inside a dynamically mapped task group is no longer skipped before the task group expands. Previously such a task was evaluated on its not-yet-expanded summary task instance, found no successful upstreams (because the upstream had already expanded away from ``map_index -1``), and was wrongly skipped instead of expanding and running.
diff --git a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
index 54d5a4d9309..fddfe97baed 100644
--- a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
+++ b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
@@ -37,7 +37,7 @@ if TYPE_CHECKING:
from airflow.models.taskinstance import TaskInstance
from airflow.serialization.definitions.mappedoperator import Operator
- from airflow.serialization.definitions.taskgroup import SerializedMappedTaskGroup
+ from airflow.serialization.definitions.taskgroup import SerializedTaskGroup
from airflow.ti_deps.dep_context import DepContext
from airflow.ti_deps.deps.base_ti_dep import TIDepStatus
from airflow.typing_compat import Unpack
@@ -152,7 +152,7 @@ class TriggerRuleDep(BaseTIDep):
return get_mapped_ti_count(task, ti.run_id, session=session)
- def _iter_expansion_dependencies(task_group: SerializedMappedTaskGroup | None) -> Iterator[str]:
+ def _iter_expansion_dependencies(task_group: SerializedTaskGroup | None) -> Iterator[str]:
if is_mapped(task):
for op in task.iter_mapped_dependencies():
yield op.task_id
@@ -177,14 +177,22 @@ class TriggerRuleDep(BaseTIDep):
assert task.task_group
# Only the not-yet-expanded summary ti (map_index < 0) needs the broad
- # "depend on every upstream ti" behavior, so a fast-triggered rule
- # (ONE_SUCCESS / ONE_FAILED / ONE_DONE) does not skip it before the
- # mapped task group has expanded (see #34023). Once the ti is expanded,
- # each instance must depend on the upstream instance(s) that share its
- # map index, otherwise a single upstream failure would wrongly trigger
- # every expanded instance (see #50210).
- if is_mapped(task.task_group) and ti.map_index < 0:
- is_fast_triggered = task.trigger_rule in (TR.ONE_SUCCESS, TR.ONE_FAILED, TR.ONE_DONE)
+ # "depend on every upstream ti" behavior, so a rule that can be satisfied
+ # by a subset of upstreams (ONE_SUCCESS / ONE_FAILED / ONE_DONE /
+ # NONE_FAILED_MIN_ONE_SUCCESS) does not skip it before the mapped task
+ # group has expanded (see #34023, #39801). Once the ti is expanded, each
+ # instance must depend on the upstream instance(s) that share its map
+ # index, otherwise a single upstream failure would wrongly trigger every
+ # expanded instance (see #50210). The task may be nested in plain task
+ # groups inside the mapped one (see #39801), so check the closest mapped
+ # ancestor rather than only the immediate parent group.
+ if ti.map_index < 0 and task.get_closest_mapped_task_group() is not None:
+ is_fast_triggered = task.trigger_rule in (
+ TR.ONE_SUCCESS,
+ TR.ONE_FAILED,
+ TR.ONE_DONE,
+ TR.NONE_FAILED_MIN_ONE_SUCCESS,
+ )
if is_fast_triggered and upstream_id not in set(
_iter_expansion_dependencies(task_group=task.task_group)
):
diff --git a/airflow-core/tests/unit/models/test_mappedoperator.py b/airflow-core/tests/unit/models/test_mappedoperator.py
index e00a821e056..3539428be43 100644
--- a/airflow-core/tests/unit/models/test_mappedoperator.py
+++ b/airflow-core/tests/unit/models/test_mappedoperator.py
@@ -1709,6 +1709,85 @@ def test_one_failed_trigger_rule_runs_on_indirect_failure_in_mapped_task_group(d
assert states["deliver_records.handle_failed_delivery"] == {0: "success", 1: "success", 2: "success"}
+def test_none_failed_min_one_success_trigger_rule_expands_in_mapped_task_group(dag_maker):
+ """Regression test for #39801.
+
+ A task with the ``NONE_FAILED_MIN_ONE_SUCCESS`` trigger rule inside a dynamically
+ expanded task group must expand and run once its upstream (also inside the group)
+ succeeds, instead of being skipped prematurely at its unexpanded (map_index -1)
+ summary ti before the task group has expanded.
+ """
+ with dag_maker(dag_id="test_none_failed_min_one_success_in_mapped_task_group") as dag:
+
+ @task
+ def init():
+ return ["seize", "the", "day"]
+
+ @task_group(group_id="tg")
+ def tg(message):
+ @task
+ def imsleepy(message):
+ return message
+
+ @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
+ def imawake(message):
+ pass
+
+ imawake(imsleepy(message))
+
+ tg.expand(message=init())
+
+ dr = dag.test()
+
+ states: dict[str, dict[int, str | None]] = defaultdict(dict)
+ for ti in dr.get_task_instances():
+ states[ti.task_id][ti.map_index] = ti.state
+
+ assert states["tg.imsleepy"] == {0: "success", 1: "success", 2: "success"}
+ assert states["tg.imawake"] == {0: "success", 1: "success", 2: "success"}
+
+
+def test_none_failed_min_one_success_trigger_rule_expands_in_nested_task_group(dag_maker):
+ """Regression test for #39801, nested-group shape.
+
+ Same as above, but the ``NONE_FAILED_MIN_ONE_SUCCESS`` task lives in a plain task
+ group nested inside the mapped one (the exact shape of the issue's reproducer), so
+ its immediate ``task_group`` is not mapped — only an ancestor is.
+ """
+ with dag_maker(dag_id="test_none_failed_min_one_success_in_nested_task_group") as dag:
+
+ @task
+ def init():
+ return ["seize", "the", "day"]
+
+ @task_group(group_id="tg")
+ def tg(message):
+ @task
+ def imsleepy(message):
+ return message
+
+ @task_group(group_id="inner")
+ def inner(message):
+ @task(trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS)
+ def imawake(message):
+ pass
+
+ imawake(message)
+
+ inner(imsleepy(message))
+
+ tg.expand(message=init())
+
+ dr = dag.test()
+
+ states: dict[str, dict[int, str | None]] = defaultdict(dict)
+ for ti in dr.get_task_instances():
+ states[ti.task_id][ti.map_index] = ti.state
+
+ assert states["tg.imsleepy"] == {0: "success", 1: "success", 2: "success"}
+ assert states["tg.inner.imawake"] == {0: "success", 1: "success", 2: "success"}
+
+
def test_mapped_operator_retry_delay_default(dag_maker):
"""
Test that MappedOperator.retry_delay returns default value when not explicitly set.