apache / apache/airflow

TriggerDagRunOperator.execute cannot be called outside of the Task Runner!

Open
#56,204 11 comments 1 reaction 1 assignee Claimed by @ephraimbuddy View on GitHub
affected_version:3.0 affected_version:main_branch area:core kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 10h
Merged PRs (30d)
483

Description

### Apache Airflow version

Other Airflow 2 version (please specify below)

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

3.0.6

### What happened?

The TriggerDagRunOperator can not be executed from a task anymore

### What you think should happen instead?

Either it should work like in Airflow 2.11 or there should be some documentation how to migrate existing DAGs

### How to reproduce

I have added an example with two minimal Dags that reproduce the problem
[x_task_run_demo.py](https://github.com/user-attachments/files/22595855/x_task_run_demo.py)

```python
import datetime
import logging
import pendulum
import pytz

from airflow import DAG
from airflow.sdk import task, Param
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator

# This DAG should schedule the other one multiple times
with DAG( dag_id = "Task-Run-Demo"
, schedule = "0 0 * * *" # Daily at midnight (00:00)
, catchup = False
, params = { "schedule_date" : Param(f"{datetime.date.today()}", type="string", format="date") }
) as dag:

@task
def schedule_dags(**kwargs):
for i in range(5):
execution_date = datetime.datetime.utcnow() + datetime.timedelta(minutes = 5)
logging.info(f"Trigger report for counter #{i} at '{execution_date}'")

try:
TriggerDagRunOperator( task_id = f"create_reports_dag_{i}"
, trigger_dag_id = "Create-Reports"
, conf = { "counter": i }
, logical_date = execution_date.replace(tzinfo = pytz.UTC),
).execute(context = kwargs)

except Exception as ex:
error_message = str(ex)
logging.error(f"Failed to trigger DAG #{i}: {error_message}")

schedule_dags()

# This DAG should not be executed directly but triggered by the Task-Run-Demo DAG
with DAG( dag_id = "Create-Reports"
, schedule = None
, catchup = False
, params = { "counter" : None }
) as dag:

@task.python
def report(params: dict):
counter = params["counter"]
logging.info(f"Current counter: {counter}")

report()

```

### Operating System

Ubuntu 24.04

### Versions of Apache Airflow Providers

_No response_

### Deployment

Docker-Compose

### Deployment details

It does not matter if I run it in docker compose or on OpenShift. The error is the same.

### Anything else?

_No response_

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.