[GitHub] [airflow] RNHTTR commented on a change in pull request #14321: Fix bug allowing task instances to survive when dagrun_timeout is exceeded

[GitBox <gi...@apache.org>]

RNHTTR commented on a change in pull request #14321:
URL: https://github.com/apache/airflow/pull/14321#discussion_r582006481



##########
File path: airflow/jobs/scheduler_job.py
##########
@@ -1696,10 +1696,18 @@ def _schedule_dag_run(
             and dag.dagrun_timeout
             and dag_run.start_date < timezone.utcnow() - dag.dagrun_timeout
         ):
-            dag_run.state = State.FAILED
-            dag_run.end_date = timezone.utcnow()
-            self.log.info("Run %s of %s has timed-out", dag_run.run_id, dag_run.dag_id)
+            dag_run.set_state(State.FAILED)
+            unfinished_task_instances = (
+                session.query(TI)
+                .filter(TI.dag_id == dag_run.dag_id)
+                .filter(TI.execution_date == dag_run.execution_date)
+                .filter(TI.state.in_(State.unfinished))
+            )
+            for task_instance in unfinished_task_instances:
+                task_instance.state = State.SKIPPED

Review comment:
       I believe setting the task instance state to failed will have the same effect. It just has to be a state in `State.finished`. I chose skipped, because I figured the tasks didn’t even get a chance to run. 




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org