timeout on Dynamic Task mapping : skipped inner task , task status is still success

Apache Airflow version

2.8.1

If "Other Airflow 2 version" selected, which one?

No response

What happened?

this is confusing and not obvious to find the problem when just looking at the grid view on a complex dag

What you think should happen instead?

maybe we should mark the task as skipped or failed

How to reproduce

from datetime import timedelta

from airflow.operators.python import PythonOperator
from pendulum import today
from airflow import DAG

dag = DAG(
    dag_id="test_dag",
    schedule_interval=None,
    dagrun_timeout=timedelta(seconds=5),
    start_date=today("UTC").add(days=-1)
)

with dag:
    def something(arg):
        import time
        time.sleep(int(arg))

    PythonOperator.partial(
        task_id="toto",
        python_callable=something,
    ).expand(op_args=[("10",), ("5",), ("4",)])

Operating System

ubuntu 22.04

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

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

The problem here is that you have a 5-second timeout on the DAG. Not sure why it didn't log but I'll investigate.

By definition, skipped is a success state see https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html#dag-run-status and dag should be marked success but because it timeout, it marked it as failed.

It's actually there that it has timed out. Please confirm

I don''t understand , what is your question ?

--

also the problem is not present without dynamic task mapping

from datetime import timedelta
from airflow.operators.python import PythonOperator
from pendulum import today
from airflow import DAG

dag = DAG(
    dag_id="test_dag",
    schedule_interval=None,
    dagrun_timeout=timedelta(seconds=5),
    start_date=today("UTC").add(days=-1)
)

with dag:
    def something(arg):
        import time
        time.sleep(int(arg))

    PythonOperator(
        task_id="toto",
        python_callable=something,
        op_args="10")
    PythonOperator(
        task_id="tata",
        python_callable=something,
        op_args="10")
``

This issue has been automatically marked as stale because it has been open for 14 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.

@ephraimbuddy

@ephraimbuddy could you please remove the tag wait for response , thanks

@ephraimbuddy

hey @ephraimbuddy , could you please update the tags of the issue , thanks

@raphaelauv

@raphaelauv

airflow/airflow/jobs/scheduler_job_runner.py

Lines 1444 to 1455 in 9901a06

    
           dag_run.set_state(DagRunState.FAILED) 
        
           unfinished_task_instances = session.scalars( 
        
               select(TI) 
        
               .where(TI.dag_id == dag_run.dag_id) 
        
               .where(TI.run_id == dag_run.run_id) 
        
               .where(TI.state.in_(State.unfinished)) 
        
           ) 
        
           for task_instance in unfinished_task_instances: 
        
               task_instance.state = TaskInstanceState.SKIPPED 
        
               session.merge(task_instance) 
        
           session.flush() 
        
           self.log.info("Run %s of %s has timed-out", dag_run.run_id, dag_run.dag_id)

.
The dagrun timeout you have on your dag causes the dag to fail. You can read it on the link above

yes the dag_run is in fail it's good , the issue is not there

the issue is that the task status is not in fail ( which is the case when not using DTM ), so if I have 10 DTM tasks I can't know where a task was skipped

This issue has been automatically marked as stale because it has been open for 14 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.

no stale

I've managed to encounter a weird variation of this issue without the timeout definition:
In one of my expanded tasks, most of the inner tasks are marked as successful, but some inner tasks fail yet marked as skipped. Then, the task itself is marked as successful.
I'll try to reproduce it when I have a chance - I'm currently experiencing the dynamic task mapping feature in its fullest to understand its current issues and limitations, and hopefully I could come up with a plan to start mitigating them.

This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.

still present on airflow 3.1.5

from datetime import timedelta
from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from pendulum import today


with DAG(
    dag_id="test_dag",
    schedule=None,
    dagrun_timeout=timedelta(seconds=5),
    start_date=today("UTC").add(days=-1)
):
    def something(arg_1, arg_2):
        import time
        if arg_2 == "titi":
            time.sleep(int(arg_1))


    for i in ["tata", "tutu", "toto", "titi"]:
        PythonOperator.partial(
            task_id=i,
            python_callable=something,
        ).expand(op_args=[("10", i), ("5", i), ("4", i)])

Navigation Menu

Repository navigation

timeout on Dynamic Task mapping : skipped inner task , task status is still success #37332

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

16 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	dag_run.set_state(DagRunState.FAILED)
	unfinished_task_instances = session.scalars(
	select(TI)
	.where(TI.dag_id == dag_run.dag_id)
	.where(TI.run_id == dag_run.run_id)
	.where(TI.state.in_(State.unfinished))
	)
	for task_instance in unfinished_task_instances:
	task_instance.state = TaskInstanceState.SKIPPED
	session.merge(task_instance)
	session.flush()
	self.log.info("Run %s of %s has timed-out", dag_run.run_id, dag_run.dag_id)

timeout on Dynamic Task mapping : skipped inner task , task status is still success #37332

Description

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

ephraimbuddy commented on Mar 25, 2024

ephraimbuddy commented on Mar 25, 2024

raphaelauv commented on Mar 25, 2024

github-actions commented on Apr 9, 2024

raphaelauv commented on Apr 9, 2024

raphaelauv commented on Jun 24, 2024

ephraimbuddy commented on Jun 24, 2024

raphaelauv commented on Jun 24, 2024

16 remaining items

github-actions commented on Nov 9, 2024

raphaelauv commented on Nov 9, 2024

shahar1 commented on Nov 19, 2024

github-actions commented on Dec 28, 2025

raphaelauv commented on Jan 5, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions