repl-chris commented on code in PR #23690:
URL: https://github.com/apache/airflow/pull/23690#discussion_r875010438
##########
airflow/executors/celery_executor.py:
##########
@@ -343,20 +346,53 @@ def _check_for_stalled_adopted_tasks(self):
# already finished, then it will be removed from this list -- so
# the only time it's still in this list is when it a) never made it
# to celery in the first place (i.e. race condition somewhere in
- # the dying executor) or b) a really long celery queue and it just
+ # the dying executor), b) celery lost the task before execution
+ # started, or c) a really long celery queue and it just
# hasn't started yet -- better cancel it and let the scheduler
# re-queue rather than have this task risk stalling for ever
timedout_keys.append(key)
- if timedout_keys:
- self.log.error(
- "Adopted tasks were still pending after %s, assuming they
never made it to celery and "
- "clearing:\n\t%s",
- self.task_adoption_timeout,
- "\n\t".join(repr(x) for x in timedout_keys),
+ if not timedout_keys:
+ return
+
+ self.log.error(
+ "Tasks were still pending after configured timeout (adopted: %s,
all: %s), "
+ "assuming they never made it to celery and clearing:\n\t%s",
+ self.task_adoption_timeout,
+ self.stalled_task_timeout,
+ "\n\t".join(repr(x) for x in timedout_keys),
+ )
+
+ try:
+ filter_for_tis = TaskInstance.filter_for_tis(timedout_keys)
+ session.query(TaskInstance).filter(
+ filter_for_tis,
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ ).update(
+ {
+ TaskInstance.state: State.SCHEDULED,
+ TaskInstance.queued_dttm: None,
+ TaskInstance.queued_by_job_id: None,
+ TaskInstance.external_executor_id: None,
+ },
+ synchronize_session=False,
)
- for key in timedout_keys:
- self.change_state(key, State.FAILED)
+ session.commit()
+ except Exception:
+ self.log.exception("Error clearing stalled tasks")
+ session.rollback()
+ return
+
+ for key in timedout_keys:
+ self.stalled_task_timeouts.pop(key, None)
+ self.running.discard(key)
+ celery_async_result = self.tasks.pop(key, None)
+ if celery_async_result:
+ try:
+ app.control.revoke(celery_async_result.task_id)
+ except Exception:
+ self.log.exception("Error revoking task instance %s from
celery", key)
Review Comment:
I've never seen it raise an exception, but looking at the celery source it
seems like it will throw if there are connectivity issues to redis....in which
case yeah a redis outage could cause tasks to "stall", and then subsequently
cause a lot of log noise on revoke. I will update this to not log the
stacktrace.
--
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]
