milton0825 commented on a change in pull request #6760: [AIRFLOW-6157] Separate
out common protocol for executors.
URL: https://github.com/apache/airflow/pull/6760#discussion_r356192360
##########
File path: airflow/executors/base_executor.py
##########
@@ -109,39 +231,24 @@ def queue_task_instance(
priority=task_instance.task.priority_weight_total,
queue=task_instance.task.queue)
- def has_task(self, task_instance: TaskInstance) -> bool:
- """
- Checks if a task is either queued or running in this executor.
-
- :param task_instance: TaskInstance
- :return: True if the task is known to this executor
- """
- return task_instance.key in self.queued_tasks or task_instance.key in
self.running
-
- def sync(self) -> None:
- """
- Sync will get called periodically by the heartbeat method.
- Executors should override this to perform gather statuses.
- """
-
def heartbeat(self) -> None:
"""
Heartbeat sent to trigger new jobs.
"""
- if not self.parallelism:
- open_slots = len(self.queued_tasks)
+ if not self._parallelism:
+ open_slots = len(self._queued_tasks)
else:
- open_slots = self.parallelism - len(self.running)
+ open_slots = self._parallelism - len(self._running)
- num_running_tasks = len(self.running)
- num_queued_tasks = len(self.queued_tasks)
+ num_running_tasks = len(self._running)
+ num_queued_tasks = len(self._queued_tasks)
self.log.debug("%s running task instances", num_running_tasks)
self.log.debug("%s in queue", num_queued_tasks)
self.log.debug("%s open slots", open_slots)
Stats.gauge('executor.open_slots', open_slots)
- Stats.gauge('executor.queued_tasks', num_queued_tasks)
+ Stats.gauge('executor._queued_tasks', num_queued_tasks)
Review comment:
I think this is not backward compatible
----------------------------------------------------------------
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:
[email protected]
With regards,
Apache Git Services
