From 6efcd18655d7b90b1ae3b4296f4975ffe7803fa2 Mon Sep 17 00:00:00 2001 From: Malthe Borch Date: Thu, 3 Feb 2022 11:42:36 +0100 Subject: [PATCH] Add additional information when queuing fails (running or queued already) --- airflow/executors/base_executor.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/airflow/executors/base_executor.py b/airflow/executors/base_executor.py index f63d3bb628961..f8ecec6f9904d 100644 --- a/airflow/executors/base_executor.py +++ b/airflow/executors/base_executor.py @@ -78,11 +78,15 @@ def queue_command( queue: Optional[str] = None, ): """Queues command to task""" - if task_instance.key not in self.queued_tasks and task_instance.key not in self.running: + is_queued = task_instance.key in self.queued_tasks + is_running = task_instance.key in self.running + if is_queued or is_running: + self.log.error( + "could not queue task %s (queued: %s; running: %s)", task_instance.key, is_queued, is_running + ) + else: self.log.info("Adding to queue: %s", command) self.queued_tasks[task_instance.key] = (command, priority, queue, task_instance) - else: - self.log.error("could not queue task %s", task_instance.key) def queue_task_instance( self,