TriggerDagRunOperator and ExternalTaskSensor with dynamic execution_date

[prithvi-git]

Trying to trigger one dag multiple times with different configs using TriggerDagRunOperator and ExternalTaskSensor. However, TriggerDagRunOperator takes parent DAGs execution_date (logical_date) for execution and that just reruns same instance of triggered DAG instead of running new instance with new config. Sample code below.
Tried passing execution_date if different ways to TriggerDagRunOperator and ExternalTaskSensor but no luck.
I'm expecting a solution so I can trigger same DAG with different config.

DAG - exec_master

@task_group(group_id='refresh_pre-prod')
def refresh_pre_prod():
prod_to_pre_prod = TriggerDagRunOperator (
task_id='prod_to_pre_prod',
trigger_dag_id="util_clone_bq_env",
execution_date='{{ dag_run.logical_date }}',
conf={"src_project_id":"production",
"trg_project_id":"pre-production"},
reset_dag_run = True)

    prod_to_pre_prod_sensor = ExternalTaskSensor(
        task_id='prod_to_pre_prod_sensor',
        external_dag_id='util_clone_bq_env',
        external_task_id='notify_completion',
        allowed_states=["success"],
        failed_states=["failed", "skipped", "upstream_failed"])

    prod_to_pre_prod >> prod_to_pre_prod_sensor

@task_group(group_id='refresh_demo')
def refresh_demo():
    prod_to_demo = TriggerDagRunOperator(
        task_id='prod_to_demo',
        trigger_dag_id="util_clone_bq_env",
        execution_date='{{ dag_run.logical_date }}',
        conf={"src_project_id":"production",
                "trg_project_id":"demo1"},
        reset_dag_run = True)

    prod_to_demo_sensor = ExternalTaskSensor(
        task_id='prod_to_demo_sensor',
        external_dag_id='util_clone_bq_env',
        external_task_id='notify_completion',
        allowed_states=["success"],
        failed_states=["failed", "skipped", "upstream_failed"])

    prod_to_demo >> prod_to_demo_sensor

refresh_pre_prod() >> bq_refresh_demo()

Thanks,
Prithvi.

[josh-fell]

Hi @prithvi-git,

IIUC your main problem that the "prod_to_demo_sensor" task doesn't properly find the DAG run that's triggered by the "prod_to_demo"? If so, you could always have the "prod_to_demo" task continue to run until the triggered DAG is completed by setting wait_for_completion=True on the TriggerDagRunOperator task. That operator also has both allowed_states and failed_states too.

[josh-fell]

Have you tried just having the TriggerDagRunOperator wait for completion?

... you could always have the "prod_to_demo" task continue to run until the triggered DAG is completed by setting wait_for_completion=True on the TriggerDagRunOperator task.

[prithvi-git]

@josh-fell
"wait_for_completion=True" didn't help. prod_to_demo still triggers the same instance triggered for prod_to_pre_prod :(
Unless we are able to pass distinct logical_date in TriggerDagRunOperator and able to use them in respective ExternalTaskSensor tasks, it will not trigger other instance of triggered DAG (util_clone_bq_env)

[prithvi-git]

AIRFLOW_CTX_EXECUTION_DATE=2023-12-29T07:39:42.683753+00:00
It's same for both triggers, hence running same instance twice.

[raunakchopra]

Were you able to get around this issue by any chance? I am also here, struggling with the Poke forever issue..

[prithvi-git]

No.. This issue is not resolved yet. I had to create multiple DAGs with different names but exact same code and call them.