apply_defaults ignored with BaseOperator._set_relatives dag assignment

[dimberman]

Ticket was created 16/Feb/17 02:13

Description

When a task is instantiated, the apply_defaults decorator retrieves default_args and params from the task's dag's defaults_args and params if a dag is specified.

During set_upstream/set_downstream if task A has a dag assigned but task B does not, task B will be assigned the dag from task A.

The set_upstream/set_downstream implicit dag assignment occurs after apply_defaults has been processed so the task will not receive the dag's default args.

bad_arg_dag.py
import datetime
import airflow.models
from airflow.operators.dummy_operator import DummyOperator

DAG = airflow.models.DAG(
dag_id='test_dag',
schedule_interval=None,
start_date=datetime.datetime(2017, 2, 14),
default_args={'owner': 'airflow', 'queue': 'some_queue'},
)

TASK1 = DummyOperator(
task_id='task1',
dag=DAG,
)

TASK2 = DummyOperator(
task_id='task2',
)
TASK2.set_upstream(TASK1)

In this case, both TASK1 and TASK2 will be assigned to DAG and TASK1 will receive the dag default queue of 'some_queue' but TASK2 will receive the airflow configuration default queue of 'default'

Use case / motivation

Related Issues

Moved here from https://issues.apache.org/jira/browse/AIRFLOW-879