ephraimbuddy commented on code in PR #30375:
URL: https://github.com/apache/airflow/pull/30375#discussion_r1160594959
##########
airflow/jobs/scheduler_job.py:
##########
@@ -1408,6 +1443,51 @@ def _send_sla_callbacks_to_processor(self, dag: DAG) ->
None:
)
self.executor.send_callback(request)
+ @provide_session
+ def _fail_tasks_stuck_in_queued(self, session: Session = NEW_SESSION) ->
None:
+ """
+ Mark tasks stuck in queued for longer than `task_queued_timeout` as
failed.
+ Tasks can get stuck in queued for a wide variety of reasons (e.g.
celery loses
+ track of a task, a cluster can't further scale up its workers, etc.),
but tasks
+ should not be stuck in queued for a long time. This will mark tasks
stuck in
+ queued for longer than `self._task_queued_timeout` as failed. If the
task has
+ available retries, it will be retried.
+ """
+ self.log.debug("Calling SchedulerJob._fail_tasks_stuck_in_queued
method")
+ for attempt in run_with_db_retries(logger=self.log):
+ with attempt:
+ self.log.debug(
+ "Running SchedulerJob.adopt_or_reset_orphaned_tasks with
retries. Try %d of %d",
+ attempt.retry_state.attempt_number,
+ MAX_DB_RETRIES,
+ )
+ try:
+ query = session.query(TI).filter(
+ TI.state == State.QUEUED,
+ TI.queued_dttm < (timezone.utcnow() -
timedelta(seconds=self._task_queued_timeout)),
+ )
+ tasks_stuck_in_queued: list[TaskInstance] = with_row_locks(
+ query, of=TI, session=session,
**skip_locked(session=session)
+ ).all()
+ tasks_stuck_in_queued_messages = []
+ for ti in tasks_stuck_in_queued:
+ tasks_stuck_in_queued_messages.append(repr(ti))
+ msg = (
+ "TI %s was in the queued state for longer than
%s.",
+ repr(ti),
+ self._task_queued_timeout,
+ )
+ ti.handle_failure(error=msg, session=session)
Review Comment:
When a task is marked as failed externally, failure callbacks are not called
https://github.com/apache/airflow/blob/0b83f064e524272e8fc1dfcd38edd7ad39bc6660/airflow/models/taskinstance.py#L1468-L1470.
My suggestion is that these task instances should not be marked failed here
instead we can collect the task instances that have overstayed and send them to
the executor and have the executor purge them from the running tasks. The
failure would now be more natural when they come back to the scheduler. We
would get a message like `the executor reports that the task instance failed
but it's still queued`, that way failure callback and retries will still work.
And I think that's the plan here but don't know why we called the failure
callback
--
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: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
