Clearing using ExternalTaskMarker will not activate external DagRuns

[eejbyfeldt]

Apache Airflow version:

2.0.1

What happened:

When clearing task across dags using ExternalTaskMarker the dag state of the external DagRun is not set to active. So cleared tasks in the external dag will not automatically start if the DagRun is a Failed or Succeeded state.

What you expected to happen:

The external DagRun run should also be set to Running state.

How to reproduce it:

Clear tasks in an external dag using an ExternalTaskMarker.

Anything else we need to know:

Looking at the code is has:

airflow/airflow/models/dag.py

Lines 1323 to 1335 in b23fc13

	if do_it:
	clear_task_instances(
	tis,
	session,
	dag=self,
	activate_dag_runs=False, # We will set DagRun state later.
	)
	self.set_dag_runs_state(
	session=session,
	start_date=start_date,
	end_date=end_date,
	state=dag_run_state,
	)

It seems like it intentionally calls the dag member method set_dag_run_state instead of letting the helper function clear_task_insntances set the DagRun state. But the member method will only change the state of DagRuns of dag where the original task is, while I believe clear_task_instances would correctly change the state of all involved DagRuns.

[zachliu]

oh no, very sad to see this 🐛
once this is fixed 🤞 i'll be able to use the ExternalTaskSensor and ExternalTaskMarker in the same DAG to achieve something like a "task reference" which allows me to control multiple parts in different DAGs using another DAG.

[yuqian90]

Does anyone know why this code needs to call self.set_dag_runs_state separately? If we simply remove the self.set_dag_runs_state call and leave activate_dag_runs as the default True, things will just work as expected.

 if do_it: 
     clear_task_instances( 
         tis, 
         session, 
         dag=self, 
         activate_dag_runs=False,  # We will set DagRun state later. 
     ) 
     self.set_dag_runs_state( 
         session=session, 
         start_date=start_date, 
         end_date=end_date, 
         state=dag_run_state, 
     ) 

[zachliu]

@yuqian90 the fastest way to get an answer is to submit a PR to it
this issue hasn't been assigned to anyone nor to a milestone, seems to be low priority
submitting a PR could definitely raise some awareness and speed things up 😁