amoghrajesh commented on code in PR #75263:
URL: https://github.com/apache/airflow/pull/75263#discussion_r4375825147
##########
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:
Not blocking, but I have an improvement suggestion here: what if you guard
inside `scheduler_dag_bag.get_dag_for_run`? That would fix this type of bug
everywhere instead of one call location at a time (it's this location this
time, but could be another some other time).
But that changes a shared helper, so not so important.
##########
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
Review Comment:
No test coverage for this branch, check a similar test:
`test_schedule_all_dag_runs_reraises_db_errors`
##########
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:
Umm, we starve one `(dag_id, task_id)` pair. Every other task of the same
broken DAG still tries to load it, and `log.exception` prints a full stack for
each, every loop. Adding to `starved_dags` instead would log once and not
create such noise
--
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]
