[GitHub] [airflow] kpathak13 commented on a change in pull request #6237: AIRFLOW-5581: Join KubernetesJobWatcher in terminate call and unblock queues from blocking forever

[GitBox <gi...@apache.org>]

kpathak13 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_r331660938
 
 

 ##########
 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:
   That should not be the case since `executor.end()` is called after exiting the executor loop.

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services