Skip to content

Commit

Permalink
[TWTTR] Don't enqueue tasks again if already queued for K8sExecutor(#60)
Browse files Browse the repository at this point in the history
Basically reverting commit 87fcc1c  and making changes specifically into the Celery Executor class only.
  • Loading branch information
msumit authored Sep 24, 2020
1 parent bac4acd commit 6162402
Show file tree
Hide file tree
Showing 3 changed files with 11 additions and 3 deletions.
4 changes: 2 additions & 2 deletions airflow/executors/base_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,9 +57,9 @@ def queue_command(self, simple_task_instance, command, priority=1, queue=None):
key = simple_task_instance.key
if key not in self.queued_tasks and key not in self.running:
self.log.info("Adding to queue: %s", command)
self.queued_tasks[key] = (command, priority, queue, simple_task_instance)
else:
self.log.info("Adding to queue even though already queued or running {}".format(command, key))
self.queued_tasks[key] = (command, priority, queue, simple_task_instance)
self.log.error("could not queue task %s", key)

def queue_task_instance(
self,
Expand Down
8 changes: 8 additions & 0 deletions airflow/executors/celery_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,14 @@ def start(self):
self._sync_parallelism
)

def queue_command(self, simple_task_instance, command, priority=1, queue=None):
key = simple_task_instance.key
if key not in self.queued_tasks and key not in self.running:
self.log.info("Adding to queue: %s", command)
else:
self.log.info("Adding to queue even though already queued or running {}".format(command, key))
self.queued_tasks[key] = (command, priority, queue, simple_task_instance)

def _num_tasks_per_send_process(self, to_send_count):
"""
How many Celery tasks should each worker process send.
Expand Down
2 changes: 1 addition & 1 deletion airflow/version.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,5 +18,5 @@
# under the License.
#

version = '1.10.4+twtr19'
version = '1.10.4+twtr21'

0 comments on commit 6162402

Please sign in to comment.