dimberman commented on a change in pull request #6237: AIRFLOW-5581: Join
KubernetesJobWatcher in terminate call and unblock queues from blocking forever
URL: https://github.com/apache/airflow/pull/6237#discussion_r331591457
##########
File path: airflow/executors/kubernetes_executor.py
##########
@@ -768,9 +787,45 @@ def _change_state(self, key, state, pod_id: str) -> None:
self.log.debug('Could not find key: %s', str(key))
self.event_buffer[key] = state
+ def _flush_task_queue(self):
+ self.log.debug('Executor shutting down, task_queue approximate
size=%d', self.task_queue.qsize())
+ while True:
+ try:
+ task = self.task_queue.get_nowait()
+ # This is a new task to run thus ok to ignore.
+ self.log.warning('Executor shutting down, will NOT run
task=%s', task)
+ self.task_queue.task_done()
+ except Empty:
+ break
+
+ def _flush_result_queue(self):
+ self.log.debug('Executor shutting down, result_queue approximate
size=%d', self.result_queue.qsize())
+ while True: # pylint: disable=too-many-nested-blocks
+ try:
+ results = self.result_queue.get_nowait()
+ self.log.warning('Executor shutting down, flushing
results=%s', results)
+ try:
+ key, state, pod_id, resource_version = results
+ self.log.info('Changing state of %s to %s :
resource_version=%d', results, state,
+ resource_version)
+ try:
+ self._change_state(key, state, pod_id)
+ except Exception as e: # pylint: disable=broad-except
+ self.log.exception('Ignoring exception: %s when
attempting to change state of %s '
+ 'to %s.', e, results, state)
+ finally:
+ self.result_queue.task_done()
+ except Empty:
+ break
+
def end(self):
"""Called when the executor shuts down"""
self.log.info('Shutting down Kubernetes executor')
+ self.log.debug('Flushing task_queue...')
+ self._flush_task_queue()
Review comment:
Is there any chance of a race condition where it continues putting in new
tasks after the queue has been flushed?
----------------------------------------------------------------
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
