Parse Error 410 in kubernetes Watcher and return latest resource version

airflow/executors/kubernetes_executor.py

@@ -25,6 +25,7 @@
 import functools
 import json
 import multiprocessing
+import re
 import time
 from queue import Empty, Queue  # pylint: disable=unused-import
 from typing import Any, Dict, List, Optional, Tuple
@@ -173,11 +174,17 @@ def process_error(self, event: Any) -> str:
         self.log.error('Encountered Error response from k8s list namespaced pod stream => %s', event)
         raw_object = event['raw_object']
         if raw_object['code'] == 410:
-            self.log.info(
-                'Kubernetes resource version is too old, must reset to 0 => %s', (raw_object['message'],)
+            message = raw_object['message']
+            latest_resource_version = self._parse_too_old_failure(message)
+            if latest_resource_version is None:
+                # Return resource version 0
+                return '0'
+            self.log.warning(
+                "Updated to new resource version: %s due to 'too old' error: %s",
+                latest_resource_version,
+                raw_object,
             )
-            # Return resource version 0
-            return '0'
+            return latest_resource_version
         raise AirflowException(
             'Kubernetes failure for %s with code %s and message: %s'
             % (raw_object['reason'], raw_object['code'], raw_object['message'])
@@ -218,6 +225,25 @@ def process_status(
                 resource_version,
             )
 
+    def _parse_too_old_failure(self, message):
+        """
+        Parse stream watcher 410 'too old resource version' error
+        to get the latest resource version from the message
+
+        See https://github.com/kubernetes-client/python/issues/609
+        """
+        regex = r"too old resource version: .* \((.*)\)"
+        result = re.search(regex, message)
+        if result is None:
+            return None
+        match = result.group(1)
+        if match is None:
+            return None
+        try:
+            return match
+        except (ValueError, TypeError):
+            return None
+
 
 class AirflowKubernetesScheduler(LoggingMixin):
     """Airflow Scheduler for Kubernetes"""

tests/executors/test_kubernetes_executor.py

@@ -507,3 +507,19 @@ def test_process_status_catchall(self):
 
         self._run()
         self.watcher.watcher_queue.put.assert_not_called()
+
+    @mock.patch.object(KubernetesJobWatcher, '_parse_too_old_failure')
+    def test_process_error_event(self, mock_too_old_failure):
+        message = "too old resource version: 27272 (43334)"
+        mock_too_old_failure.return_value = '43334'
+        self.pod.status.phase = 'Pending'
+        self.pod.metadata.resource_version = '43334'
+        raw_object = {"code": 410, "message": message}
+        self.events.append({"type": "ERROR", "object": self.pod, "raw_object": raw_object})
+        self._run()
+        mock_too_old_failure.assert_called_once_with(message)
+
+    def test_parse_too_old_failure(self):
+        message = "too old resource version: 27272 (43334)"
+        latest_version = self.watcher._parse_too_old_failure(message)
+        assert latest_version == '43334'