DAG in queued when it has Deferrable

[dinjazelena98]

Apache Airflow version

3.0.2

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

No response

What happened?

Hi, i have an Airflow Operator which uses self.defer() to call an Deferrable Trigger. Inside that deferrable trigger we are just waiting for event to happen. Once event happens it yields TriggerEvent back to the worker and executing "method_name" from self.defer() method. Here i want to trigger next DAG which needs that event, and go back to deferring. Now next DAG lasts for much longer, and i want to have possible concurrent runs.

But when ever next DAG is triggered, my initial DAG goes to status "queued". I absolutely cant figure out why.

    def execute(self, context: dict[str, Any]) -> None:
        self.defer(
            trigger=DeferrableTriggerClass(**params),
            method_name="trigger",
        )

    def trigger(self, context: dict[str, Any], event: dict[str, Any]) -> None:
        TriggerDagRunOperator(
            task_id="__trigger",
            trigger_dag_id="next_dag",
            conf={event["target"]},
            wait_for_completion=False,
        ).execute(context)
        
        self.defer(
            trigger=DeferrableTriggerClass(**params),
            method_name="trigger",
        )

First i tried something like above. But it seems that after calling TriggerDagRunOperator, actual task gets done and anything after it never gets executed. So initially my idea was to have an single DAG that is always either triggering the DAG or waiting for event to happen. But after using TriggerDagRunOperator, nothing gets executed after it, and DAG is considered success.

Then to mitigate problem above, i tried to just make this DAG run as schedule="@continuous", so after every time it gets event, trigger the DAG with that event, and then based on schedule to create a new run. But then once the the DAG that was waiting for event, triggers the next DAG with TriggerDagRunOperator, DAG is marked green, and by schedule creates a new instance of the DAG, but it gets to queued for the runtime of the DAG that triggered.

I tested this without having deferrable task, and it works good, i have initial DAG that always triggers next one and does not have any dependency on it.

Why does it get queued? Or why is DAG automatically finished after calling TriggerDagRun inside method of Operator.

What you think should happen instead?

Once the DAG that has deferrable triggers the next DAG, it should be retriggered because of @continuous schedule. But it goes to queued for the runtime of the DAG that it has triggered.

How to reproduce

Have a DAG with self.defer() trigger and make it trigger some other DAG in method_name that is triggered once we move back to worker from triggerer

Operating System

Ubuntu

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

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

This looks more like a discussion - hopefully someone will be able to help

[dinjazelena98]

Hm, not sure to be honest, its not behaving as expected, but ok

[potiuk]

Not sure either. That's why discussion, rather than issue where you clearly show error and know it's a problem is better.

[dinjazelena98]

@potiuk , hey i have been trying to solve it for a week now, checked everything, trigger logs, metadata database, scheduler logs. Absolutely no idea why it gets stuck in queued for the entire runtime of the DAG that it triggered, here is simple snippet to reproduce:

import asyncio
import pendulum
import time

from typing import Any, AsyncIterator

from airflow.sdkimport dag
from airflow.operators.bash import BashOperator
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.triggers.base import BaseTrigger, TriggerEvent
from airflow.models.baseoperator import BaseOperator


class DummySleepTrigger(BaseTrigger):
    def __init__(self, sleep_time: int):
        super().__init__()
        self.sleep_time = sleep_time

    def serialize(self) -> tuple[str, dict[str, Any]]:
        return ("dags.dummy.DummySleepTrigger", {"sleep_time": self.sleep_time})

    async def run(self) -> AsyncIterator[TriggerEvent]:
        self.log.info(f"Dummy trigger started, sleeping for {self.sleep_time} seconds.")
        await asyncio.sleep(self.sleep_time)
        self.log.info("Dummy trigger finished sleeping.")
        yield TriggerEvent({"status": "success", "dummy_value": f"slept for {self.sleep_time}"})


class DummySleepOperator(BaseOperator):
    def __init__(self, sleep_time: int, **kwargs) -> None:
        super().__init__(**kwargs)
        self.sleep_time = sleep_time

    def execute(self, context):
        self.defer(trigger=DummySleepTrigger(sleep_time=self.sleep_time), method_name="resumed_method")

    def resumed_method(self, context, event=None):
        self.log.info(f"Resumed! Event details: {event}")
        # Return a dummy dictionary, just like the real operator
        return {"status": "success", "dummy_data": [1, 2, 3]}


@dag(
    dag_id="ingestion_dummy_dag",
    start_date=pendulum.datetime(2025, 1, 1),
)
def ingestion_dummy_dag():
    BashOperator(
        task_id="process_data_dummy",
        bash_command="echo '--- Starting dummy ingestion work ---' && sleep 60 && echo '--- Dummy ingestion work finished ---'",
    )

ingestion_dummy_dag()


@dag(
    dag_id="monitor_dummy_dag",
    start_date=pendulum.datetime(2025, 1, 1),
    schedule="@continuous",
    max_active_runs=1,
)
def monitor_dummy_dag():
    monitor_task = DummySleepOperator(task_id="dummy_monitor_task", sleep_time=10)

    trigger_ingestion_task = TriggerDagRunOperator(
        task_id="trigger_ingestion_dummy_dag",
        trigger_dag_id="ingestion_dummy_dag",
        wait_for_completion=False,
        conf=monitor_task.output
    )

    monitor_task >> trigger_ingestion_task

monitor_dummy_dag()

So looking at this, it should have monitor DAG that defers (wait for events, here it sleeps just for showcasing), once its done, scheduler picks up the TriggerEvent, calls method_name, and then we use trigger operator to trigger next DAG. Once the ingestion DAG is triggered, the monitor should be again in defer state sleeping, but actually it gets queued, and stays there for the entire runtime of the ingestion DAG.

I tested without defer, and its works, so problem is not in TriggerDagRunOperator.

Can u please try it?