I have a dependencies list of task ids [['A','B'],['C','D']], which should result in the following stream:
A >> B
C >> D
It generally works except from the case [['A','B']['C','B']], which should results in both operators A and C ending and than executing B. Here is how the dependecies are created:
for dep in dependencies_list:
for i in range(len(dep)-1):
task_id_update = "update_bq_ts_{}".format(dep[i])
task_id_start = "start_df_pipeline_{}".format(dep[i+1])
last_ops[task_id_update] >> first_ops[task_id_start]
The dictionaries last_ops and first_ops are key:str (task_id) value:Operator.
The B operator (start_df_pipeline_ *) definition looks like this:
def start_pipeline(dag, pipeline_name, runtime_args):
return CloudDataFusionStartPipelineOperator(
task_id='start_df_pipeline_{pipeline_name}',
instance_name=instance_name,
location=location,
pipeline_name=pipeline_name,
runtime_args=runtime_args,
trigger_rule="all_success"
)
In the graph view it shows the dependecies correctly:
While if I open the Task instance details of strat_df_pipeline_* task the upstream list is empty
And when I run the whole DAG it executes the start_df_pipeline_* not waiting for the two update_bq_* to finish.

