This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 3b627816506 [v3-3-test] Stop skipping none_failed_min_one_success
tasks in mapped task groups (#69377) (#70318)
3b627816506 is described below
commit 3b62781650608946ed1fe5a2ac5e03c8a66c687c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jul 24 14:53:57 2026 +0530
[v3-3-test] Stop skipping none_failed_min_one_success tasks in mapped task
groups (#69377) (#70318)
(cherry picked from commit dd029a899537c61709226af4d4b0f514fb1c025c)
Co-authored-by: Shahar Epstein <[email protected]>
---
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.
