abhijeets25012-tech commented on code in PR #75263:
URL: https://github.com/apache/airflow/pull/75263#discussion_r4380326802
##########
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:
Thanks @amoghrajesh and @ephraimbuddy. Starving the run is the better fix.
I will pass `starved_dag_runs` into `_task_concurrency_allows_execution` and
add `(dag_id, run_id)` there. That way the run is filtered by the existing
`_build_schedulable_tis_query`, counted by `found_new_filters`, and the
deserialization failure is logged once per run instead of once per task and
scheduler loop.
I will also add the test you described: two runs of the same Dag with
`max_active_tis_per_dag=1`, where `get_dag_for_run` fails only for the older
run; the newer run's TI should be queued.
I will update this PR rather than opening a new one.
##########
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:
Review Comment:
Thanks @amoghrajesh. I agree that guarding inside
`scheduler_dag_bag.get_dag_for_run` would cover every caller instead of one
call site.
I kept this localized because it is the direct fix for the reported crash
and is easier to backport to 3.4. I am happy to follow up with a separate
issue/PR to centralize the handling in the shared helper once this is merged,
unless you would prefer it in this PR.
--
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]
