ephraimbuddy commented on code in PR #75263:
URL: https://github.com/apache/airflow/pull/75263#discussion_r4375988157
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -603,10 +603,19 @@ def _task_concurrency_allows_execution(
task_id = task_instance.task_id
run_id = task_instance.run_id
- serialized_dag = self.scheduler_dag_bag.get_dag_for_run(
- dag_run=task_instance.dag_run,
- session=session,
- )
+ try:
+ serialized_dag = self.scheduler_dag_bag.get_dag_for_run(
+ dag_run=task_instance.dag_run,
+ session=session,
+ )
+ except DBAPIError:
+ raise
+ except Exception:
+ self.log.exception(
+ "Failed to load Dag %s for task instance %s, skipping it",
dag_id, task_instance
+ )
+ starved_tasks.add((dag_id, task_id))
Review Comment:
Yeah, I think we should starve the run instead. Pass `starved_dag_runs` into
`_task_concurrency_allows_execution` and add `(dag_id, run_id)` here.
`_build_schedulable_tis_query` already filters on it and `found_new_filters`
already counts it. It also keeps the broken run's other tasks out of the refill
queries, where each one re-deserializes and logs a traceback, since `DBDagBag`
doesn't cache failed loads.
A test with two runs of the same task, `max_tis=1`, and `get_dag_for_run`
failing only for the older run would cover it: the newer run's TI should be
queued.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
