Airflow is trying to schedule tasks prior to DAG's start_date

[a-chernykh]

Apache Airflow version

2.2.2

What happened

I have a DAG which start_date was 09/01/2021 initially but then it was changed to 11/01/2021. This DAG has some runs prior to 11/01/2021 that did not get a chance to finish. I can now see that scheduler is still trying to schedule the runs prior to 11/01/2021. But none of the tasks in these runs are starting because of the start_date check I presume. This is maxing out the active runs thus blocking any other days within the DAG range to be scheduled.

What you expected to happen

Scheduler should not be trying to schedule runs that are prior to DAG's start_date.

How to reproduce

No response

Operating System

Airflow Docker

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

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

[raphaelauv]

The scheduler is not going to delete dag runs already existing , so you have to manually delete them , I think it's safe this way

[a-chernykh]

This is fine and it actually makes sense that the scheduler won't delete the old runs, however, I don't think it is the right behavior for the scheduler to continue scheduling runs that are earlier than the actual start date of the DAG.

[avkirilishin]

It's very strange because in 2.2.2 we have strong checks on max_active_runs and don't create more DagRuns than this value. And I cannot reproduce it in the current main.

When were these runs created? Did you upgrade Airflow after it?

[potiuk]

Is this the same as #21011 @uranusjr ? WDYT?

[a-chernykh]

@avkirilishin runs were created in 2.2.0 or 2.2.1 likely then Airflow was upgraded to 2.2.2.

[avkirilishin]

@avkirilishin runs were created in 2.2.0 or 2.2.1 likely then Airflow was upgraded to 2.2.2.

@andreychernih I think there are two different problems:

1. The problem is related to the different logic of the scheduler before and after the update. Maybe there are no tasks in the running dags or something else. Can you show the rows for this dag run in dag, dag_run and task_instance?

2. I agree with you that it is not the right behavior for the scheduler to continue scheduling runs that are earlier than the actual start date of the DAG or Task. It can happen, for example, after turning the dag off and back on. So I made a PR to fix it: Add dependency to the running_deps #21684

[uranusjr]

(Sorry I missed this) I think #21011 is different. That one schedules the task (incorrectly) at start_date even if that time does not lie on the schedule, and the fix is to delay the first run to a time after start_date that matches the schedule. The problem description here says however the tun is scheduled before start_date.

[ephraimbuddy]

@avkirilishin runs were created in 2.2.0 or 2.2.1 likely then Airflow was upgraded to 2.2.2.

@andreychernih I think there are two different problems:

1. The problem is related to the different logic of the scheduler before and after the update. Maybe there are no tasks in the running dags or something else. Can you show the rows for this dag run in dag, dag_run and task_instance?
2. I agree with you that it is not the right behavior for the scheduler to continue scheduling runs that are earlier than the actual start date of the DAG or Task. It can happen, for example, after turning the dag off and back on. So I made a PR to fix it: Add dependency to the running_deps #21684

Since tasks go through queued state before moving to running, I think that the scheduler has the check here:

airflow/airflow/ti_deps/dependencies_deps.py

Line 90 in 3c12c2e

ExecDateAfterStartDateDep(),

?

[avkirilishin]

Since tasks go through queued state before moving to running, I think that the scheduler has the check here:

airflow/airflow/ti_deps/dependencies_deps.py

Line 90 in 3c12c2e

ExecDateAfterStartDateDep(),

?

I think SCHEDULER_QUEUED_DEPS is actually not equivalent to the logic in the scheduler:

airflow/airflow/ti_deps/dependencies_deps.py

Lines 67 to 75 in 3c4524b

	# TODO(aoen): SCHEDULER_QUEUED_DEPS is not coupled to actual scheduling/execution
	# in any way and could easily be modified or removed from the scheduler causing
	# this dependency to become outdated and incorrect. This coupling should be created
	# (e.g. via a dag_deps analog of ti_deps that will be used in the scheduler code,
	# or allow batch deps checks) to ensure that the logic here is equivalent to the logic
	# in the scheduler.
	# Right now there's one discrepancy between this context and how scheduler schedule tasks:
	# Scheduler will check if the executor has the task instance--it is not possible
	# to check the executor outside scheduler main process.

[ephraimbuddy]

What I think we should do but I have not tried any:

1. Make sure that we don't create dagruns prior to a DAG's start date
   or
2. Exclude the task instances here:

   airflow/airflow/jobs/scheduler_job.py

   Lines 280 to 289 in f0bbb9d

   	query = (
   	session.query(TI)
   	.join(TI.dag_run)
   	.filter(DR.run_type != DagRunType.BACKFILL_JOB, DR.state == DagRunState.RUNNING)
   	.join(TI.dag_model)
   	.filter(not_(DM.is_paused))
   	.filter(TI.state == TaskInstanceState.SCHEDULED)
   	.options(selectinload('dag_model'))
   	.order_by(-TI.priority_weight, DR.execution_date)
   	)

I think the solution you have right now, will have the scheduler put those task instances in queued state and never move them to running state which might not be a very good experience for users. WDYT @avkirilishin

[ashb]

Can someone please give me exact reproduction steps for this including a dag I can run?

[github-actions]

This issue has been automatically marked as stale because it has been open for 30 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.

[github-actions]

This issue has been closed because it has not received response from the issue author.