Triggering DAG with Future Date - clarification

[trlopes1974]

Hi.
According to airflow documentation:
Triggering DAG with Future Date
If you want to use ‘external trigger’ to run future-dated data intervals, set allow_trigger_in_future = True in scheduler section in airflow.cfg. This only has effect if your DAG is defined with schedule=None. When set to False (the default value), if you manually trigger a run with future-dated data intervals, the scheduler will not execute it until its data_interval_start is in the past

I have 2 dags.
The first has one task that uses the triggerdagrunoperator to schedule the 2nd dag into 3 future datetime dag runs.

IF allow_trigger_in_future = True and I use triggerdagrunoperator with an execution_date in the future, ALL 3 dag runs execute immediately ( desppite that schedule=None in the 2nd dag )

According to the stated above if allow_trigger_in_future = True and Schedule=None, shouldn't the dag run on the pecified execution_date ???

Note: after I set allow_trigger_in_future = False the dag runs on the specified date/time

example:
DAG1:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
from airflow.models.param import Param

import logging

logger = logging.getLogger(__name__)

doc_md_DAG = """

## Dag de teste para ver se agendamentos funcionam
## Owner
> Tiago.Lopes
"""

default_args = {
    'owner' : 'ttauto',
    'depends_on_past': False,
    'start_date': datetime(2021, 1, 1),
    'email': ['tt.auto@kyndryl.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay' : timedelta(seconds=20),
    'max_retry_delay' : timedelta(seconds=60),
    'retry_exponential_backoff': True    
}

A001_future_dag = DAG(dag_id='A001_future_dag',
    default_args=default_args,
    catchup=False,
    schedule_interval=None, 
    schedule=None,
    start_date=datetime(2024, 2, 5),
    doc_md=doc_md_DAG,
    description='testes',
    max_active_tasks=1,
    max_active_runs=1,
    render_template_as_native_obj=True,
    tags=['TESTE'] 
    
) 

#task de preparação 
def task_runner(**context):
    import sys,os
    from datetime import datetime, timezone, timedelta
    from airflow.operators.trigger_dagrun import TriggerDagRunOperator
    logger.info("********** task_runner Task Starting! ***********")
    try:
        autodagname="run_A002_teste_future_dag_run_for"
        lisbon_tz = timezone(timedelta(hours=1), "Europe/Lisbon")
        
        current_time = datetime.now()
        for i in range(1,4):
            adjusted_time = current_time + timedelta(minutes=i*15)
            adjusted_time_end = adjusted_time + timedelta(minutes=10)
            adjusted_time.replace(tzinfo=lisbon_tz)            
            adjusted_time_end.replace(tzinfo=lisbon_tz)  
            logger.debug(f"setting date to start:{adjusted_time.isoformat()}  end:{adjusted_time_end.isoformat()}")            
            trigger = TriggerDagRunOperator(    
                    task_id=f"run_A002_teste_future_dag_run_for_",
                    trigger_dag_id='A002_teste_future_dag_run',
                    execution_date=adjusted_time.isoformat(),
                    #conf={"start_date" : adjusted_time.isoformat(),"end_date" : adjusted_time_end.isoformat(),"data_interval_start" : adjusted_time.isoformat(),"data_interval_end" : adjusted_time_end.isoformat() },              
                    wait_for_completion=False   )
            logger.info(f"A calendarizar o dag:{autodagname} para start:{adjusted_time.isoformat()}  end:{adjusted_time_end.isoformat()}")
            trigger.execute(context)
    except Exception as e:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        errmsg = f">>exc_obj: {exc_obj} ,exc_type: {exc_type} , fname: {fname} , exc_tb.tb_lineno: {str(exc_tb.tb_lineno)} " 
        logger.error(f">>exc_obj: {exc_obj} ,exc_type: {exc_type} , fname: {fname} , exc_tb.tb_lineno: {str(exc_tb.tb_lineno)} " )
        logger.info("********** Prepare Task Finished! ***********")
    
task001_task_runner = PythonOperator(task_id='task_runner', python_callable=task_runner, provide_context=True, dag=A001_future_dag)


#workflow
task001_task_runner

DAG2:

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta

import logging
logger = logging.getLogger(__name__)

doc_md_DAG = """
## Dag de teste para ver se agendamentos funcionam
## Owner
> Tiago.Lopes

"""

default_args = {
    'owner' : 'ttauto',
    'depends_on_past': False,
    'start_date': datetime(2024, 2, 5),
    'email': ['tt.auto@kyndryl.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0  
}


A002_teste_future_dag_run = DAG(dag_id='A002_teste_future_dag_run',
    default_args=default_args,
    catchup=False,
    schedule_interval=None, 
    schedule=None,
    start_date=datetime(2024, 2, 5),
    doc_md=doc_md_DAG,
    description='testes',
    max_active_tasks=1,
    max_active_runs=3,
    render_template_as_native_obj=True,
    tags=['TESTE'] 
    
) 


#task de preparação 
def schedule_dag_run(**context):
    from datetime import datetime
    logger.info("********** schedule_dag_run Task Starting! ***********")
    dgr = context['dag_run']
    logger.debug(f"dgr runid:{dgr.run_id}, execution_date:{dgr.execution_date}, start_date:{dgr.start_date }, external_trigger:{dgr.external_trigger }, run_type:{dgr.run_type }, queued_at:{dgr.queued_at }   ")
    logger.debug(f"Executei em:  {datetime.now()}")
    logger.info("********** schedule_dag_run Task Finished! ***********")
task_executa = PythonOperator(task_id='schedule_dag_run', python_callable=schedule_dag_run, provide_context=True, dag=A002_teste_future_dag_run)

task_executa 

[rahulpateltr]

@trlopes1974, have you got any response/ any solution for the above question? we are also facing similar issue with MWAA airflow. we are using 2.8.1 version on mwaa instance.
@trlopes1974 trlopes1974

[trlopes1974]

Well, I did make it work.
with this:
after I set allow_trigger_in_future = False the dag runs on the specified date/time (it makes no sense to me, but it works )

Allow externally triggered DagRuns for Execution Dates in the future

#allow_trigger_in_future = True
allow_trigger_in_future = False

this is the dag that runs in the future:
default_args = {
'owner' : 'ttauto',
'depends_on_past': False,
'start_date': datetime(2024, 2, 5),
'email': ['tt.auto@somewhere.com'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 0
}

A002_teste_future_dag_run = DAG(dag_id='A002_teste_future_dag_run',
default_args=default_args,
catchup=False,
schedule_interval=None,
schedule=None,
start_date=datetime(2024, 2, 5),
doc_md=doc_md_DAG,
description='testes',
max_active_tasks=1,
max_active_runs=3,
render_template_as_native_obj=True,
tags=['TESTE']

)

We also implementes other solutions like using sensor and deploy a task in te future, as the DateTimeSensor is async, it does not occupy a slot in he pool.