Short circuit task in expanded task group fails when it returns false

[luederm]

Apache Airflow version

2.6.1

What happened

I have a short circuit task which is in a task group that is expanded. The task work correctly when it returns true but the task fails when it returns false with the following error:

sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) insert or update on table "xcom" violates foreign key constraint "xcom_task_instance_fkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(pipeline_output_to_s3, transfer_output_file.already_in_manifest_short_circuit, manual__2023-05-24T20:21:35.420606+00:00, -1) is not present in table "task_instance".

It looks like it sets the map-index to -1 when false is returned which is causing the issue.
If one task fails this way in the the task group, all other mapped tasks fail as well, even if the short circuit returns true.
When this error occurs, all subsequent DAG runs will be stuck indefinitely in the running state unless the DAG is deleted.

What you think should happen instead

Returning false from the short circuit operator should skip downstream tasks without affecting other task groups / map indexes or subsequent DAG runs.

How to reproduce

include a short circuit operator in a mapped task group and have it return false.

Operating System

Red Hat Enterprise Linux 7

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

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

[hussein-awala]

@luederm Thank you for reporting this issue. For the future issues, it would be greatly appreciated if you could provide a reproducible example similar to the original one. This would help us identify whether the problem is due to misconfiguration/misuse of the operator or a bug.

For this one, I can reproduce it with:

import pendulum

from airflow.decorators import dag, task, task_group, short_circuit_task
from airflow.operators.python import ShortCircuitOperator

@dag(
    dag_id="test_short_circuit",
    start_date=pendulum.datetime(2023, 1, 1),
    schedule_interval=None,
)
def dag():
    @task
    def t1():
        return [1, 2, 3, 4]

    @task_group(group_id="task_group")
    def group(x):
        s = ShortCircuitOperator(
            task_id="short_circuit",
            python_callable=lambda arg: arg % 2 == 0,
            op_kwargs={"arg": x},
        )

        @task
        def t2(arg):
            return arg

        s >> t2(x)

    group.expand(x=t1())

dag()

[luederm]

@hussein-awala Noted. Thanks for looking into this so quickly.