anzinmhd opened a new pull request, #76323:
URL: https://github.com/apache/airflow/pull/76323
When a mapped task instance fails in a Dag with `fail_fast=True`, other
instances of the same mapped task can continue running or start afterward. This
PR identifies the triggering task instance by its unique execution ID so its
mapped siblings receive the existing fail-fast handling.
### Problem
Mapped task instances share a `task_id`, but represent separate executions.
For example:
- `mapped_work[0]` fails.
- `mapped_work[1]` is already running.
- `mapped_work[2]` is waiting for capacity.
With `fail_fast=True`, the running sibling should be marked failed and the
waiting sibling should be skipped.
Instead, the running sibling continues, and the waiting sibling can start
after the first instance fails. This allows additional work to execute despite
the Dag’s fail-fast setting.
### Cause
`_stop_remaining_tasks()` excludes the triggering instance using:
```python
ti.task_id == task_instance.task_id
```
Every mapped instance of the same task has that task ID. Consequently, the
condition excludes all mapped siblings from failure handling, rather than
excluding only the execution that triggered it.
Those siblings never reach the existing logic that marks running instances
failed and other unfinished instances skipped.
### Fix
Replace the task-definition ID comparison with a unique task-instance ID
comparison:
```python
ti.id == task_instance.id
```
This distinguishes the specific triggering execution from its mapped
siblings.
The helper now:
- Leaves the triggering instance’s failure transition to its caller.
- Processes other mapped instances, even though they share the same
`task_id`.
- Preserves already-successful and already-failed instances.
- Retains the existing teardown-task handling.
The identity check works for both mapped and unmapped instances, without
separate branches based on map index.
It also does not depend on the triggering instance already being marked
failed. This matters because the Execution API can invoke fail-fast handling
before executing that instance’s final state update.
The production change is limited to this comparison. No API schema, database
migration, or dependency changes are required.
### Reproduction
Use the following Dag with capacity for two concurrent task instances:
```python
import time
from airflow.sdk import DAG, task
with DAG(
dag_id="reproduce_fail_fast_mapped_siblings",
schedule=None,
catchup=False,
fail_fast=True,
max_active_tasks=2,
max_active_runs=1,
):
@task(retries=0)
def mapped_work(item: int) -> int:
if item == 0:
time.sleep(10)
raise ValueError("Intentional failure to exercise fail-fast
handling.")
time.sleep(120)
return item
mapped_work.expand(item=[0, 1, 2])
```
1. Unpause the Dag and trigger a new run.
2. Open the mapped task-instance list.
3. Confirm that indexes 0 and 1 are running while index 2 waits for capacity.
4. Wait for index 0 to fail after approximately 10 seconds.
5. Observe the remaining instances without manually changing their states.
**Before:** Index 1 continues running, and index 2 can start after index 0
fails.
**After:** Index 1 is marked failed and index 2 is skipped.
Use a new Dag run for each comparison. Sufficient executor and pool capacity
is needed to have indexes 0 and 1 running together before the intentional
failure.
### Version verification
- Reproduced with a live Dag on the local development `main` checkout.
- Before-and-after recordings demonstrate the failure and corrected behavior.
- Affected beta and released versions have not been independently
established.
### Tests and validation
Added eight database-backed regression cases:
- Six model-level cases covering running, scheduled, and queued mapped
siblings, with the triggering instance either still running or already failed.
- Two Execution API cases that send a failure update through the real
endpoint without mocking `_stop_remaining_tasks()`.
The model tests also verify that the triggering instance and
already-successful or already-failed mapped instances are preserved.
Validation completed:
- All eight new regression cases fail with the original task-ID comparison.
- All eight pass with the task-instance ID comparison.
- The latest focused verification run passed **41 tests**, including the new
cases and existing fail-fast, scheduler heartbeat-timeout, and retry coverage.
- Automated verification ran inside Breeze using isolated SQLite test
databases.
- Targeted Ruff formatting and lint checks pass.
- Live before-and-after reproduction completed.
The full CI and database-backend matrix have not been run locally.
### Before
Index 0 fails, but its mapped siblings continue executing. Index 2 can start
after the failure despite `fail_fast=True`.
https://github.com/user-attachments/assets/10971825-74c1-4856-9f64-cf7c09ba0b7f
### After
After index 0 fails, the running mapped sibling is marked failed and the
waiting mapped sibling is skipped.
https://github.com/user-attachments/assets/1471b774-b2f3-4a65-97da-37220dee0468
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Codex (GPT-6)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
