Compatibility with kazoo-2.9.0 (#2428)

Now the select() method may raise `TypeError` and `IOError` exceptions if the socket is closed.

patroni/dcs/zookeeper.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import select
+import six
 import time
 
 from kazoo.client import KazooClient, KazooState, KazooRetry
@@ -50,11 +51,21 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
         return super(PatroniSequentialThreadingHandler, self).create_connection(*args, **kwargs)
 
     def select(self, *args, **kwargs):
-        """Python3 raises `ValueError` if socket is closed, because fd == -1"""
+        """
+        Python 3.XY may raise following exceptions if select/poll are called with an invalid socket:
+        - `ValueError`: because fd == -1
+        - `TypeError`: Invalid file descriptor: -1 (starting from kazoo 2.9)
+        Python 2.7 may raise the `IOError` instead of `socket.error` (starting from kazoo 2.9)
+
+        When it is appropriate we map these exceptions to `socket.error`.
+        """
+
         try:
             return super(PatroniSequentialThreadingHandler, self).select(*args, **kwargs)
-        except ValueError as e:
-            raise select.error(9, str(e))
+        except IOError as e:
+            raise (select.error(e.errno, e.strerror) if six.PY2 else e)
+        except (TypeError, ValueError) as e:
+            raise (e if six.PY2 and isinstance(e, TypeError) else select.error(9, str(e)))
 
 
 class PatroniKazooClient(KazooClient):

tests/test_zookeeper.py

@@ -124,9 +124,11 @@ class TestPatroniSequentialThreadingHandler(unittest.TestCase):
         self.assertIsNotNone(self.handler.create_connection((), 40))
         self.assertIsNotNone(self.handler.create_connection(timeout=40))
 
-    @patch.object(SequentialThreadingHandler, 'select', Mock(side_effect=ValueError))
     def test_select(self):
-        self.assertRaises(select.error, self.handler.select)
+        with patch.object(SequentialThreadingHandler, 'select', Mock(side_effect=ValueError)):
+            self.assertRaises(select.error, self.handler.select)
+        with patch.object(SequentialThreadingHandler, 'select', Mock(side_effect=IOError)):
+            self.assertRaises(Exception, self.handler.select)
 
 
 class TestPatroniKazooClient(unittest.TestCase):