leonsmith commented on a change in pull request #14861:
URL: https://github.com/apache/airflow/pull/14861#discussion_r596377787
##########
File path: airflow/executors/base_executor.py
##########
@@ -182,10 +182,10 @@ def trigger_tasks(self, open_slots: int) -> None:
sorted_queue = self.order_queued_tasks_by_priority()
for _ in range(min((open_slots, len(self.queued_tasks)))):
- key, (command, _, _, ti) = sorted_queue.pop(0)
+ key, (command, _, queue, ti) = sorted_queue.pop(0)
self.queued_tasks.pop(key)
self.running.add(key)
- self.execute_async(key=key, command=command, queue=None,
executor_config=ti.executor_config)
+ self.execute_async(key=key, command=command, queue=queue,
executor_config=ti.executor_config)
Review comment:
Yep celery is not broken as it does override the `trigger_tasks` to pass
the queue.
The BaseExecutor should expose all the functionality (like it used to)
though without having to have executors override this function? Otherwise the
queue argument to `execute_async` is redundant?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
