DAG Run fails when chaining multiple empty mapped tasks

[park-peter]

Apache Airflow version

2.4.3

What happened

A significant fraction of the DAG Runs of a DAG that has 2+ consecutive mapped tasks which are are being passed an empty list are marked as failed when all tasks are either succeeding or being skipped. This was supposedly fixed with issue #25200 but the problem still persists.

What you think should happen instead

The DAG Run should be marked success.

How to reproduce

The real world version of this DAG has several mapped tasks that all point to the same list, and that list is frequently empty. I have made a minimal reproducible example.

from datetime import datetime

from airflow import DAG
from airflow.decorators import task


with DAG(dag_id="break_mapping", start_date=datetime(2022, 3, 4)) as dag:

    @task
    def add_one(x: int):
        return x + 1

    @task
    def say_hi():
        print("Hi")

    @task
    def say_bye():
        print("Bye")

    added_values = add_one.expand(x=[])
    added_more_values = add_one.expand(x=[])
    added_more_more_values = add_one.expand(x=[])
    say_hi() >> say_bye() >> added_values
    added_values >> added_more_values >> added_more_more_values

Operating System

Debian Bullseye

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[uranusjr]

Does changing the task’s trigger rule make a difference? I think the default rule is to require all upstreams to succeed, and having an empty upstream is not considered successful.

[ashb]

The issue is not the tasks being skipped, but that some of the runs fail.

@park-peter Can you look through the scheduler logs for the runs that fail? See if anything in particular jumps out?

[park-peter]

[2022-11-22 17:28:42,114] {scheduler_job.py:536} INFO - Sending TaskInstanceKey(dag_id='break_mapping', task_id='say_bye', run_id='scheduled__2022-06-15T00:00:00+00:00', try_number=1, map_index=-1) to executor with priority 4 and queue default
[2022-11-22 17:28:42,114] {base_executor.py:95} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'break_mapping', 'say_bye', 'scheduled__2022-06-15T00:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/example-dag.py']
[2022-11-22 17:28:42,115] {local_executor.py:81} INFO - QueuedLocalWorker running ['airflow', 'tasks', 'run', 'break_mapping', 'say_bye', 'scheduled__2022-06-15T00:00:00+00:00', '--local', '--subdir', 'DAGS_FOLDER/example-dag.py']
[2022-11-22 17:28:42,117] {scheduler_job.py:588} INFO - Executor reports execution of break_mapping.say_hi run_id=scheduled__2022-06-15T00:00:00+00:00 exited with status success for try_number 1
[2022-11-22 17:28:42,119] {scheduler_job.py:631} INFO - TaskInstance Finished: dag_id=break_mapping, task_id=say_hi, run_id=scheduled__2022-06-15T00:00:00+00:00, map_index=-1, run_start_date=2022-11-22 17:28:41.097404+00:00, run_end_date=2022-11-22 17:28:41.220139+00:00, run_duration=0.122735, state=success, executor_state=success, try_number=1, max_tries=0, job_id=196, pool=default_pool, queue=default, priority_weight=5, operator=_PythonDecoratedOperator, queued_dttm=2022-11-22 17:28:40.980893+00:00, queued_by_job_id=148, pid=418
[2022-11-22 17:28:42,202] {task_command.py:376} INFO - Running <TaskInstance: break_mapping.say_bye scheduled__2022-06-15T00:00:00+00:00 [queued]> on host 5fbf19be21aa
[2022-11-22 17:28:42,556] {mappedoperator.py:666} INFO - Marking <TaskInstance: break_mapping.add_one scheduled__2022-06-15T00:00:00+00:00 [None]> as SKIPPED since the map has 0 values to expand
[2022-11-22 17:28:42,565] {dagrun.py:616} ERROR - Deadlock; marking run <DagRun break_mapping @ 2022-06-15 00:00:00+00:00: scheduled__2022-06-15T00:00:00+00:00, state:running, queued_at: 2022-11-22 17:28:40.930510+00:00. externally triggered: False> failed
[2022-11-22 17:28:42,565] {dagrun.py:644} INFO - DagRun Finished: dag_id=break_mapping, execution_date=2022-06-15 00:00:00+00:00, run_id=scheduled__2022-06-15T00:00:00+00:00, run_start_date=2022-11-22 17:28:40.952076+00:00, run_end_date=2022-11-22 17:28:42.565779+00:00, run_duration=1.613703, state=failed, external_trigger=False, run_type=scheduled, data_interval_start=2022-06-15 00:00:00+00:00, data_interval_end=2022-06-16 00:00:00+00:00, dag_hash=51298338f12d384f99c852836833fa98

[ephraimbuddy]

Looks like this is all we need:

diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index 9e02f4775f..f3fc068b04 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -716,6 +716,7 @@ class DagRun(Base, LoggingMixin):
             # During expansion we may change some tis into non-schedulable
             # states, so we need to re-compute.
             if expansion_happened:
+                changed_tis = True
                 new_unfinished_tis = [t for t in unfinished_tis if t.state in State.unfinished]
                 finished_tis.extend(t for t in unfinished_tis if t.state in State.finished)
                 unfinished_tis = new_unfinished_tis

in addition to @ashb's previous fix