Clearing a running dag run does not reset the queued_at timestamp

[ferruzzi]

Apache Airflow version

3.1.3

If "Other Airflow 2/3 version" selected, which one?

No response

What happened?

When you clear a dag run, it changes the state to QUEUED which should set the queued_at timestamp to the newly-re-queued time, but for some reason it isn't. I submitted #59066 but that only fixed it when the run is in SUCCESS or FAILURE state, if you clear a RUNNING run, it still doesn't work. The discussion in that PR also implies that there might be something deeper going on with SQLAlchemy which my "fix" may possibly just be plastering over instead of actually fixing, and that may be worth a look.

I might get to this one, but I am trying to wrap a few other things up before the holidays, so if someone gets to it before me, that would be great.

What you think should happen instead?

When you clear a dag run, it changes the state to QUEUED which should set the queued_at timestamp to the newly-re-queued time

How to reproduce

I tested this by running a dag like this:

with DAG(dag_id="really_long_dag"):
    BashOperator(task_id='sleep_task', bash_command='sleep 10000')

Run the dag and check the queued_at time in the database. You can use psql in the Breeze environment or perhaps your IDE has a database monitoring connection; I know PyCharm does.

Clear the run (I used the UI) and the run state will flash though QUEUED and to RUNNING, but the queued_at timestamp will remain unchanged.

Wait for the dag to finish or force it to a terminal state and clear it again. You'll see it flash through the state change again, but this time the queued_at time in the db will have updated.

Operating System

linux

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[henry3260]

Can I work on this?

[tirkarthi]

I tried debugging this. The docstring for the method has a note on this. The start_date, clear_number and queued_at are updated only for dagruns in finished state with dr.state in State.finished_dr_states being false for a running dagrun.

airflow/airflow-core/src/airflow/models/taskinstance.py

Lines 200 to 206 in 7218cf0

	Clear a set of task instances, but make sure the running ones get killed.
	Also sets Dagrun's `state` to QUEUED and `start_date` to the time of execution.
	But only for finished DRs (SUCCESS and FAILED).
	Doesn't clear DR's `state` and `start_date`for running
	DRs (QUEUED and RUNNING) because clearing the state for already
	running DR is redundant and clearing `start_date` affects DR's duration.

airflow/airflow-core/src/airflow/models/taskinstance.py

Lines 281 to 307 in 7218cf0

	if dr.state in State.finished_dr_states:
	dr.state = dag_run_state
	dr.start_date = timezone.utcnow()
	if run_on_latest_version:
	dr_dag = scheduler_dagbag.get_latest_version_of_dag(dr.dag_id, session=session)
	dag_version = DagVersion.get_latest_version(dr.dag_id, session=session)
	if dag_version:
	# Change the dr.created_dag_version_id so the scheduler doesn't reject this
	# version when it sets the dag_run.dag
	dr.created_dag_version_id = dag_version.id
	dr.dag = dr_dag
	dr.verify_integrity(session=session, dag_version_id=dag_version.id)
	for ti in dr.task_instances:
	ti.dag_version_id = dag_version.id
	else:
	dr_dag = scheduler_dagbag.get_dag_for_run(dag_run=dr, session=session)
	if not dr_dag:
	log.warning("No serialized dag found for dag '%s'", dr.dag_id)
	if dr_dag and not dr_dag.disable_bundle_versioning and run_on_latest_version:
	bundle_version = dr.dag_model.bundle_version
	if bundle_version is not None and run_on_latest_version:
	dr.bundle_version = bundle_version
	if dag_run_state == DagRunState.QUEUED:
	dr.last_scheduling_decision = None
	dr.start_date = None
	dr.clear_number += 1
	dr.queued_at = timezone.utcnow()

[tirkarthi]

Clearing a dagrun in running state doesn't move it to queued to be moved to running state by scheduler. The dagrun remains in running state. Related commit that made this change in Airflow 2.7.0 before which clearing will always set the dagrun state to queued. Perhaps for deadlines use case the dr.queued_at could be updated irrespective of the dagrun's current state during clearing but that's still semantically little different because dagrun never moves to queued state to update queued_at and then to later become running.

Ref commit : 070ecbd

[ferruzzi]

Hm, I wonder if that's still the desired behavior. I can do an easy deadline-specific workaround by adding a check before the if dr.state in State.finished_dr_states: but I wonder if it's something we want to change. What are the advantages of this way instead of just always flagging it QUEUED? Are we saying that if it's currently running when a user clears it, that it just restarts instead of getting tossed back into the queue? That doesn't seem like the expected behavior, but maybe I'm wrong.

[henry3260]

I suspect this change was made for performance optimizations or to simplify the state transitions. That said, I agree the current behavior feels unintuitive.

[ferruzzi]

Reading through the original Issue that spawned that change, the request was that we should not change the start_date on a clear which I agree with, but I feel like the queued_at time should reset

edit: confusing phrasing

[ferruzzi]

Started an email thread on the dev list to get community thoughts. My proposal is that we should always kick the run back to the queue and update the queued_at regardless of the current state.