DAG Run fails when chaining multiple empty mapped tasks

[collinmcnulty]

Apache Airflow version

2.3.3 (latest released)

What happened

On Kubernetes Executor and Local Executor (others not tested) a significant fraction of the DAG Runs of a DAG that has two consecutive mapped tasks which are are being passed an empty list are marked as failed when all tasks are either succeeding or being skipped.

What you think should happen instead

The DAG Run should be marked success.

How to reproduce

Run the following DAG on Kubernetes Executor or Local Executor.

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")


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

Operating System

Debian Bullseye

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==1!4.0.0
apache-airflow-providers-cncf-kubernetes==1!4.1.0
apache-airflow-providers-elasticsearch==1!4.0.0
apache-airflow-providers-ftp==1!3.0.0
apache-airflow-providers-google==1!8.1.0
apache-airflow-providers-http==1!3.0.0
apache-airflow-providers-imap==1!3.0.0
apache-airflow-providers-microsoft-azure==1!4.0.0
apache-airflow-providers-mysql==1!3.0.0
apache-airflow-providers-postgres==1!5.0.0
apache-airflow-providers-redis==1!3.0.0
apache-airflow-providers-slack==1!5.0.0
apache-airflow-providers-sqlite==1!3.0.0
apache-airflow-providers-ssh==1!3.0.0

Deployment

Astronomer

Deployment details

Local was tested on docker compose (from astro-cli)

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]

Hmm, I wonder if this has something to do with the fact that a task mapped against an empty list produces one single task instance with state REMOVED.

[ephraimbuddy]

It seems like a race condition between the scheduler calling dag_run.update states vs mappedoperator expand_mapped_task updating the task instance state to skipped

[ephraimbuddy]

Here's a solution:

diff --git a/airflow/models/mappedoperator.py b/airflow/models/mappedoperator.py
index a883ff2404..1617944013 100644
--- a/airflow/models/mappedoperator.py
+++ b/airflow/models/mappedoperator.py
@@ -665,6 +665,12 @@ class MappedOperator(AbstractOperator):
                     total_length,
                 )
                 unmapped_ti.state = TaskInstanceState.SKIPPED
+                # Skip the downstream tasks as well since they'd eventually be skipped
+                session.query(TaskInstance).filter(
+                    TaskInstance.dag_id == self.dag_id,
+                    TaskInstance.run_id == run_id,
+                    TaskInstance.task_id.in_(self.downstream_task_ids)
+                    ).update({TaskInstance.state: TaskInstanceState.SKIPPED}, synchronize_session='fetch')
             else:
                 # Otherwise convert this into the first mapped index, and create
                 # TaskInstance for other indexes.

However, I'm worried that trigger rules are not respected.

This error occurs only when the scheduler expands the mapped task through dagrun.task_instance_scheduling_decisions because the scheduler session is not yet committed when the skipping occurs, the ti ends up in unfinished_tis

[ephraimbuddy]

The error at the scheduler log:

2022-07-26 09:26:49,032] {dag.py:3084} INFO - Setting next_dagrun for break_mapping to 2022-05-21T00:00:00+00:00, run_after=2022-05-22T00:00:00+00:00
[2022-07-26 09:26:49,090] {mappedoperator.py:665} INFO - Marking <TaskInstance: break_mapping.add_one scheduled__2022-05-07T00:00:00+00:00 [None]> as SKIPPED since the map has 0 values to expand
[2022-07-26 09:26:49,113] {dagrun.py:588} ERROR - Deadlock; marking run <DagRun break_mapping @ 2022-05-07 00:00:00+00:00: scheduled__2022-05-07T00:00:00+00:00, state:running, queued_at: 2022-07-26 09:26:46.876995+00:00. externally triggered: False> failed

The task log for the successful task with failed dagrun(last lines), the two other tasks were skipped as expected:

[2022-07-26, 09:26:48 UTC] {logging_mixin.py:115} INFO - Hi
[2022-07-26, 09:26:48 UTC] {python.py:173} INFO - Done. Returned value was: None
[2022-07-26, 09:26:48 UTC] {taskinstance.py:1393} INFO - Marking task as SUCCESS. dag_id=break_mapping, task_id=say_hi, execution_date=20220507T000000, start_date=20220726T092647, end_date=20220726T092648
[2022-07-26, 09:26:49 UTC] {local_task_job.py:163} INFO - Task exited with return code 0
[2022-07-26, 09:26:49 UTC] {local_task_job.py:272} INFO - 0 downstream tasks scheduled from follow-on schedule check

The task log for the successful task with successful dagrun(last lines), the two other tasks were skipped as expected:

[2022-07-26, 09:26:43 UTC] {logging_mixin.py:115} INFO - Hi
[2022-07-26, 09:26:43 UTC] {python.py:173} INFO - Done. Returned value was: None
[2022-07-26, 09:26:43 UTC] {taskinstance.py:1393} INFO - Marking task as SUCCESS. dag_id=break_mapping, task_id=say_hi, execution_date=20220503T000000, start_date=20220726T092641, end_date=20220726T092643
[2022-07-26, 09:26:43 UTC] {local_task_job.py:163} INFO - Task exited with return code 0
[2022-07-26, 09:26:43 UTC] {mappedoperator.py:665} INFO - Marking <TaskInstance: break_mapping.add_one scheduled__2022-05-03T00:00:00+00:00 [None]> as SKIPPED since the map has 0 values to expand
[2022-07-26, 09:26:43 UTC] {local_task_job.py:272} INFO - 0 downstream tasks scheduled from follow-on schedule check

[RNHTTR]

Could this be related to #23361 and resolved by #25312 ?

[uranusjr]

I think it is related to #23361, but I’m not entirely sure if #25312 would do it. It would fix the deadlock, but the underlying race condition between the scheduler and task process still exists. I think #25060 is also likely a consequence to this race condition?

[collinmcnulty]

@uranusjr Does that mean it will be solved by #24432? That's marked as the solution to #25060 but seems distinct from this race condition.

[uranusjr]

I don’t think this is fixed yet; as mentioned in #25060 (comment):

It seems like the merged PR only solved scheduler crashes which is part of this issue but not the real issue.

[collinmcnulty]

For anyone else experiencing this, there is a workaround to put a sleep between your two sets of mapped tasks.

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(trigger_rule="all_done")
    def sleep_task():
        sleep(5)

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

[ldacey]

FYI - ran into the same issue on Airflow 2.3.4. I had to revert back to 2.3.2 for a stable experience for now, and I had to delete one row from my task instance table (a UPDATE was trying to change map_index to 0 which already existed, apparently, so the primary constraint was violated and the scheduler crashed).

In my case, none of my dynamic tasks happen to have empty lists though - there are values.