DatasetOrTimeSchedule causes deadlock

[RobbertDM]

Apache Airflow version

2.9.3

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

No response

What happened?

I want to use DatasetOrTimeSchedule as a schedule in my DAG.

When I do that and wait a day for the schedule to trigger and also the dataset trigger from another DAG, I indeed see 2 DAG runs triggered, one triggered by the time and one by the dataset (has the dataset icon as well).
Now, the problem is, the time triggered one (displayed first) is queued, and the dataset triggered on is running, but all the tasks have no status. They keep being in this state ad infinitum without anything running.

What's also weird is that the time triggered one is queued just one second after the dataset triggered one:
Queued At: 2024-08-07, 02:01:45 CEST
And the Dataset triggered one:
Queued at 2024-08-07, 02:01:44 CEST
Started 2024-08-07, 02:01:44 CEST

Which I find weird because it should've been queued at 02:00:00 CEST according to its schedule.

I should note I have max_active_runs=1 and depends_on_past=True.

What you think should happen instead?

The Dataset trigger should not influence the time schedule. They should run independently. The time schedule run should be queued and start running at its scheduled time. The tasks should actually run. There should be no deadlock.

How to reproduce

From the docs:

from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable


@dag(
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 1 * * 3", timezone="UTC"), datasets=(dag1_dataset | dag2_dataset)
    )
    # Additional arguments here, replace this comment with actual arguments
)
def example_dag():
    # DAG tasks go here
    pass

Set max_active_tasks=1 and depends_on_past=True.
Have a second DAG trigger this DAG with dag1_dataset or dag2_dataset

Operating System

Our Airflow runs on kubernetes on EKS

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

We use conveyor https://docs.conveyordata.com/technical-reference/airflow/airflow-installation-details

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]

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.

[sunank200]

I have following DAG:

with DAG(
    dag_id="dataset_produces_1",
    catchup=False,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    schedule="@daily",
    tags=["produces", "dataset-scheduled"],
) as dag1:
    # [START task_outlet]
    BashOperator(outlets=[dag1_dataset], task_id="producing_task_1", bash_command="sleep 5", retries=2)
    # [END task_outlet]

with DAG(
    dag_id="dataset_produces_2",
    catchup=False,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    schedule=None,
    tags=["produces", "dataset-scheduled"],
) as dag2:
    BashOperator(outlets=[dag2_dataset], task_id="producing_task_2", bash_command="sleep 10")

default_args = {
    'depends_on_past': True,
}
with DAG(
    dag_id="conditional_dataset_and_time_based_timetable",
    catchup=False,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("* * * * *", timezone="UTC"), datasets=(dag1_dataset & dag2_dataset)
    ),
    tags=["dataset-time-based-timetable"],
    max_active_tasks=1,
    default_args=default_args,
) as dag8:
    BashOperator(
        outlets=[Dataset("s3://dataset_time_based/dataset_other_unknown.txt")],
        task_id="conditional_dataset_and_time_based_timetable",
        bash_command="sleep 5",
    )

But this runs fine for me with the correct status for each task. I am running it locally using Breeze using a local executor. Is this correct step to replicate this @RobbertDM ?

[github-actions]

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.