Repository: zalando/patroni
Pull request #1982 (2021-06-24 08:42:43)

Compatibility with kazoo-2.7+
Old versions of `kazoo` immediately discarded all requests to Zookeeper if the connection is in the `SUSPENDED` state. This is absolutely fine because Patroni is handling retries on its own.
Starting from 2.7, kazoo started queueing requests instead of discarding and as a result, the Patroni HA  loop was getting stuck until the connection to Zookeeper is reestablished, causing no demote of the Postgres.
In order to return to the old behavior we override the `KazooClient._call()` method.

In addition to that, we ensure that the `Postgresql.reset_cluster_info_state()` method is called even if DCS failed (the order of calls was changed in the #1820).

Close https://github.com/zalando/patroni/issues/1981


Patch:
diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index fb5194cc..187d634f 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -33,13 +33,17 @@ def install_requirements(what):
 
 
 def install_packages(what):
+    from mapping import versions
+
     packages = {
         'zookeeper': ['zookeeper', 'zookeeper-bin', 'zookeeperd'],
         'consul': ['consul'],
     }
     packages['exhibitor'] = packages['zookeeper']
     packages = packages.get(what, [])
-    ver = str({'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what))
+    ver = versions.get(what)
+    subprocess.call(['sudo', 'sed', '-i', 's/pgdg main.*$/pgdg main {0}/'.format(ver),
+                     '/etc/apt/sources.list.d/pgdg.list'])
     subprocess.call(['sudo', 'apt-get', 'update', '-y'])
     return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev', 'wget'] + packages)
 
diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py
new file mode 100644
index 00000000..abe30a39
--- /dev/null
+++ b/.github/workflows/mapping.py
@@ -0,0 +1,1 @@
+versions = {'etcd': '9.6', 'etcd3': '14', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '14'}
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 30891889..3aaeada6 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -29,7 +29,7 @@ Feature: basic replication
     Then I receive a response code 200
 
   Scenario: check stuck sync replica
-    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": true, "maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
     Then I receive a response code 200
     And I create table on postgres0
     And table mytest is present on postgres1 after 2 seconds
@@ -43,7 +43,7 @@ Feature: basic replication
     Then I receive a response code 200
     When I issue a GET request to http://127.0.0.1:8010/async
     Then I receive a response code 200
-    When I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": null, "maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
     Then I receive a response code 200
     And I drop table on postgres0
 
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index ac0167c9..f3b9ac38 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -4,8 +4,9 @@ import select
 import time
 
 from kazoo.client import KazooClient, KazooState, KazooRetry
-from kazoo.exceptions import NoNodeError, NodeExistsError
+from kazoo.exceptions import NoNodeError, NodeExistsError, SessionExpiredError
 from kazoo.handlers.threading import SequentialThreadingHandler
+from kazoo.protocol.states import KeeperState
 
 from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
 from ..exceptions import DCSError
@@ -55,6 +56,20 @@ class PatroniSequentialThreadingHandler(SequentialThreadingHandler):
             raise select.error(9, str(e))
 
 
+class PatroniKazooClient(KazooClient):
+
+    def _call(self, request, async_object):
+        # Before kazoo==2.7.0 it wasn't possible to send requests to zookeeper if
+        # the connection is in the SUSPENDED state and Patroni was strongly relying on it.
+        # The https://github.com/python-zk/kazoo/pull/588 changed it, and now such requests are queued.
+        # We override the `_call()` method in order to keep the old behavior.
+
+        if self._state == KeeperState.CONNECTING:
+            async_object.set_exception(SessionExpiredError())
+            return False
+        return super(PatroniKazooClient, self)._call(request, async_object)
+
+
 class ZooKeeper(AbstractDCS):
 
     def __init__(self, config):
@@ -68,10 +83,10 @@ class ZooKeeper(AbstractDCS):
                    'cert': 'certfile', 'key': 'keyfile', 'key_password': 'keyfile_password'}
         kwargs = {v: config[k] for k, v in mapping.items() if k in config}
 
-        self._client = KazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
-                                   timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
-                                   sleep_func=time.sleep), command_retry=KazooRetry(deadline=config['retry_timeout'],
-                                   max_delay=1, max_tries=-1, sleep_func=time.sleep), **kwargs)
+        self._client = PatroniKazooClient(hosts, handler=PatroniSequentialThreadingHandler(config['retry_timeout']),
+                                          timeout=config['ttl'], connection_retry=KazooRetry(max_delay=1, max_tries=-1,
+                                          sleep_func=time.sleep), command_retry=KazooRetry(max_delay=1, max_tries=-1,
+                                          deadline=config['retry_timeout'], sleep_func=time.sleep), **kwargs)
         self._client.add_listener(self.session_listener)
 
         self._fetch_cluster = True
diff --git a/patroni/ha.py b/patroni/ha.py
index 507d586b..7d08dcae 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -417,6 +417,9 @@ class Ha(object):
         if msg:
             return msg
 
+        if not self.is_paused():
+            self.state_handler.handle_parameter_change()
+
         role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica'
         # It might happen that leader key in the standby cluster references non-exiting member.
         # In this case it is safe to continue running without changing recovery.conf
@@ -1305,8 +1308,12 @@ class Ha(object):
     def _run_cycle(self):
         dcs_failed = False
         try:
-            self.load_cluster_from_dcs()
-            self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover)
+            try:
+                self.load_cluster_from_dcs()
+                self.state_handler.reset_cluster_info_state(self.cluster, self.patroni.nofailover)
+            except Exception:
+                self.state_handler.reset_cluster_info_state(None, self.patroni.nofailover)
+                raise
 
             if self.is_paused():
                 self.watchdog.disable()
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index b33fff8b..13055dbe 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -109,10 +109,12 @@ class Postgresql(object):
         # Last known running process
         self._postmaster_proc = None
 
-        if self.is_running():
+        if self.is_running():  # we are "joining" already running postgres
             self.set_state('running')
             self.set_role('master' if self.is_leader() else 'replica')
-            self.config.write_postgresql_conf()  # we are "joining" already running postgres
+            # postpone writing postgresql.conf for 12+ because recovery parameters are not yet known
+            if self.major_version < 120000 or self.is_leader():
+                self.config.write_postgresql_conf()
             hba_saved = self.config.replace_pg_hba()
             ident_saved = self.config.replace_pg_ident()
             if hba_saved or ident_saved:
@@ -367,6 +369,17 @@ class Postgresql(object):
     def is_leader(self):
         return bool(self._cluster_info_state_get('timeline'))
 
+    def replay_paused(self):
+        return self._cluster_info_state_get('replay_paused')
+
+    def resume_wal_replay(self):
+        self._query('SELECT pg_catalog.pg_{0}_replay_resume()'.format(self.wal_name))
+
+    def handle_parameter_change(self):
+        if self.major_version >= 140000 and self.replay_paused():
+            logger.info('Resuming paused WAL replay for PostgreSQL 14+')
+            self.resume_wal_replay()
+
     def pg_control_timeline(self):
         try:
             return int(self.controldata().get("Latest checkpoint's TimeLineID"))
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index d0e5fc5b..c61a7695 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -702,7 +702,9 @@ class ConfigHandler(object):
         if wal_receiver_primary_slot_name is not None:
             self._current_recovery_params['primary_slot_name'][0] = wal_receiver_primary_slot_name
 
-        required = {'restart': 0, 'reload': 0}
+        # Increment the 'reload' to enforce write of postgresql.conf when joining the running postgres
+        required = {'restart': 0,
+                    'reload': int(not self._postgresql.cb_called and self._postgresql.major_version >= 120000)}
 
         def record_missmatch(mtype):
             required['restart' if mtype else 'reload'] += 1
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index 519b0b24..4f40f189 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -189,7 +189,9 @@ class Rewind(object):
                     need_rewind = False
                 elif master_timeline > 1:
                     cur.execute('TIMELINE_HISTORY %s', (master_timeline,))
-                    history = bytes(cur.fetchone()[1]).decode('utf-8')
+                    history = cur.fetchone()[1]
+                    if not isinstance(history, six.string_types):
+                        history = bytes(history).decode('utf-8')
                     logger.debug('master: history=%s', history)
         except Exception:
             return logger.exception('Exception when working with master via replication connection')
diff --git a/patroni/postgresql/validator.py b/patroni/postgresql/validator.py
index 93cdce54..945d5afd 100644
--- a/patroni/postgresql/validator.py
+++ b/patroni/postgresql/validator.py
@@ -108,7 +108,7 @@ parameters = CaseInsensitiveDict({
     ),
     'archive_timeout': Integer(90300, None, 0, 1073741823, 's'),
     'array_nulls': Bool(90300, None),
-    'authentication_timeout': Integer(90300, None,  1, 600, 's'),
+    'authentication_timeout': Integer(90300, None, 1, 600, 's'),
     'autovacuum': Bool(90300, None),
     'autovacuum_analyze_scale_factor': Real(90300, None, 0, 100, None),
     'autovacuum_analyze_threshold': Integer(90300, None, 0, 2147483647, None),
@@ -151,12 +151,14 @@ parameters = CaseInsensitiveDict({
         Integer(90600, None, 30, 86400, 's')
     ),
     'checkpoint_warning': Integer(90300, None, 0, 2147483647, 's'),
+    'client_connection_check_interval': Integer(140000, None, '0', '2147483647', 'ms'),
     'client_encoding': String(90300, None),
     'client_min_messages': Enum(90300, None, ('debug5', 'debug4', 'debug3', 'debug2',
                                               'debug1', 'log', 'notice', 'warning', 'error')),
     'cluster_name': String(90500, None),
     'commit_delay': Integer(90300, None, 0, 100000, None),
     'commit_siblings': Integer(90300, None, 0, 1000, None),
+    'compute_query_id': EnumBool(140000, None, ('auto',)),
     'config_file': String(90300, None),
     'constraint_exclusion': EnumBool(90300, None, ('partition',)),
     'cpu_index_tuple_cost': Real(90300, None, 0, 1.79769e+308, None),
@@ -168,6 +170,7 @@ parameters = CaseInsensitiveDict({
     'DateStyle': String(90300, None),
     'db_user_namespace': Bool(90300, None),
     'deadlock_timeout': Integer(90300, None, 1, 2147483647, 'ms'),
+    'debug_invalidate_system_caches_always': Integer(140000, None, '0', '0', None),
     'debug_pretty_print': Bool(90300, None),
     'debug_print_parse': Bool(90300, None),
     'debug_print_plan': Bool(90300, None),
@@ -176,6 +179,7 @@ parameters = CaseInsensitiveDict({
     'default_table_access_method': String(120000, None),
     'default_tablespace': String(90300, None),
     'default_text_search_config': String(90300, None),
+    'default_toast_compression': Enum(140000, None, ('pglz', 'lz4')),
     'default_transaction_deferrable': Bool(90300, None),
     'default_transaction_isolation': Enum(90300, None, ('serializable', 'repeatable read',
                                                         'read committed', 'read uncommitted')),
@@ -188,6 +192,7 @@ parameters = CaseInsensitiveDict({
     ),
     'effective_cache_size': Integer(90300, None, 1, 2147483647, '8kB'),
     'effective_io_concurrency': Integer(90300, None, 0, 1000, None),
+    'enable_async_append': Bool(140000, None),
     'enable_bitmapscan': Bool(90300, None),
     'enable_gathermerge': Bool(100000, None),
     'enable_hashagg': Bool(90300, None),
@@ -203,12 +208,14 @@ parameters = CaseInsensitiveDict({
     'enable_partition_pruning': Bool(110000, None),
     'enable_partitionwise_aggregate': Bool(110000, None),
     'enable_partitionwise_join': Bool(110000, None),
+    'enable_resultcache': Bool(140000, None),
     'enable_seqscan': Bool(90300, None),
     'enable_sort': Bool(90300, None),
     'enable_tidscan': Bool(90300, None),
     'escape_string_warning': Bool(90300, None),
     'event_source': String(90300, None),
     'exit_on_error': Bool(90300, None),
+    'extension_destdir': String(140000, None),
     'external_pid_file': String(90300, None),
     'extra_float_digits': Integer(90300, None, -15, 3, None),
     'force_parallel_mode': EnumBool(90600, None, ('regress',)),
@@ -229,8 +236,10 @@ parameters = CaseInsensitiveDict({
     'hot_standby': Bool(90300, None),
     'hot_standby_feedback': Bool(90300, None),
     'huge_pages': EnumBool(90400, None, ('try',)),
+    'huge_page_size': Integer(140000, None, '0', '2147483647', 'kB'),
     'ident_file': String(90300, None),
     'idle_in_transaction_session_timeout': Integer(90600, None, 0, 2147483647, 'ms'),
+    'idle_session_timeout': Integer(140000, None, '0', '2147483647', 'ms'),
     'ignore_checksum_failure': Bool(90300, None),
     'ignore_invalid_pages': Bool(130000, None),
     'ignore_system_indexes': Bool(90300, None),
@@ -283,6 +292,7 @@ parameters = CaseInsensitiveDict({
     'log_parameter_max_length_on_error': Integer(130000, None, -1, 1073741823, 'B'),
     'log_parser_stats': Bool(90300, None),
     'log_planner_stats': Bool(90300, None),
+    'log_recovery_conflict_waits': Bool(140000, None),
     'log_replication_commands': Bool(90500, None),
     'log_rotation_age': Integer(90300, None, 0, 35791394, 'min'),
     'log_rotation_size': Integer(90300, None, 0, 2097151, 'kB'),
@@ -336,6 +346,7 @@ parameters = CaseInsensitiveDict({
         Integer(90400, 90600, 1, 8388607, None),
         Integer(90600, None, 0, 262143, None)
     ),
+    'min_dynamic_shared_memory': Integer(140000, None, '0', '2147483647', 'MB'),
     'min_parallel_index_scan_size': Integer(100000, None, 0, 715827882, '8kB'),
     'min_parallel_relation_size': Integer(90600, 100000, 0, 715827882, '8kB'),
     'min_parallel_table_scan_size': Integer(100000, None, 0, 715827882, '8kB'),
@@ -344,7 +355,7 @@ parameters = CaseInsensitiveDict({
         Integer(100000, None, 2, 2147483647, 'MB')
     ),
     'old_snapshot_threshold': Integer(90600, None, -1, 86400, 'min'),
-    'operator_precedence_warning': Bool(90500, None),
+    'operator_precedence_warning': Bool(90500, 140000),
     'parallel_leader_participation': Bool(110000, None),
     'parallel_setup_cost': Real(90600, None, 0, 1.79769e+308, None),
     'parallel_tuple_cost': Real(90600, None, 0, 1.79769e+308, None),
@@ -358,6 +369,8 @@ parameters = CaseInsensitiveDict({
     'pre_auth_delay': Integer(90300, None, 0, 60, 's'),
     'quote_all_identifiers': Bool(90300, None),
     'random_page_cost': Real(90300, None, 0, 1.79769e+308, None),
+    'recovery_init_sync_method': Enum(140000, None, ('fsync', 'syncfs')),
+    'remove_temp_files_after_crash': Bool(140000, None),
     'replacement_sort_tuples': Integer(90600, 110000, 0, 2147483647, None),
     'restart_after_crash': Bool(90300, None),
     'row_security': Bool(90500, None),
@@ -373,6 +386,7 @@ parameters = CaseInsensitiveDict({
     'ssl_ca_file': String(90300, None),
     'ssl_cert_file': String(90300, None),
     'ssl_ciphers': String(90300, None),
+    'ssl_crl_dir': String(140000, None),
     'ssl_crl_file': String(90300, None),
     'ssl_dh_params_file': String(100000, None),
     'ssl_ecdh_curve': String(90400, None),
@@ -388,7 +402,7 @@ parameters = CaseInsensitiveDict({
     'stats_temp_directory': String(90300, None),
     'superuser_reserved_connections': (
         Integer(90300, 90600, 0, 8388607, None),
-        Integer(90600, None, 0, 262143, None),
+        Integer(90600, None, 0, 262143, None)
     ),
     'synchronize_seqscans': Bool(90300, None),
     'synchronous_commit': (
@@ -424,6 +438,7 @@ parameters = CaseInsensitiveDict({
     'track_counts': Bool(90300, None),
     'track_functions': Enum(90300, None, ('none', 'pl', 'all')),
     'track_io_timing': Bool(90300, None),
+    'track_wal_io_timing': Bool(140000, None),
     'transaction_deferrable': Bool(90300, None),
     'transaction_isolation': Enum(90300, None, ('serializable', 'repeatable read',
                                                 'read committed', 'read uncommitted')),
@@ -433,7 +448,7 @@ parameters = CaseInsensitiveDict({
     'unix_socket_group': String(90300, None),
     'unix_socket_permissions': Integer(90300, None, 0, 511, None),
     'update_process_title': Bool(90300, None),
-    'vacuum_cleanup_index_scale_factor': Real(110000, None, 0, 1e+10, None),
+    'vacuum_cleanup_index_scale_factor': Real(110000, 140000, 0, 1e+10, None),
     'vacuum_cost_delay': (
         Integer(90300, 120000, 0, 100, 'ms'),
         Real(120000, None, 0, 100, 'ms')
@@ -443,8 +458,10 @@ parameters = CaseInsensitiveDict({
     'vacuum_cost_page_hit': Integer(90300, None, 0, 10000, None),
     'vacuum_cost_page_miss': Integer(90300, None, 0, 10000, None),
     'vacuum_defer_cleanup_age': Integer(90300, None, 0, 1000000, None),
+    'vacuum_failsafe_age': Integer(140000, None, '0', '2100000000', None),
     'vacuum_freeze_min_age': Integer(90300, None, 0, 1000000000, None),
     'vacuum_freeze_table_age': Integer(90300, None, 0, 2000000000, None),
+    'vacuum_multixact_failsafe_age': Integer(140000, None, '0', '2100000000', None),
     'vacuum_multixact_freeze_min_age': Integer(90300, None, 0, 1000000000, None),
     'vacuum_multixact_freeze_table_age': Integer(90300, None, 0, 2000000000, None),
     'wal_buffers': Integer(90300, None, -1, 262143, '8kB'),


Test patch:
diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py
index fbda3999..89a9b222 100644
--- a/.github/workflows/run_tests.py
+++ b/.github/workflows/run_tests.py
@@ -23,7 +23,9 @@ def main():
 
     env = os.environ.copy()
     if sys.platform.startswith('linux'):
-        version = {'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what)
+        from mapping import versions
+
+        version = versions.get(what)
         path = '/usr/lib/postgresql/{0}/bin:.'.format(version)
         unbuffer = ['timeout', '600', 'unbuffer']
         args = ['--tags=-skip'] if what == 'etcd' else []
diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py
index 9d4d460d..fdd3e993 100644
--- a/tests/test_exhibitor.py
+++ b/tests/test_exhibitor.py
@@ -24,7 +24,7 @@ class TestExhibitor(unittest.TestCase):
 
     @patch('urllib3.PoolManager.request', Mock(return_value=urllib3.HTTPResponse(
         status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}')))
-    @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
+    @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient)
     def setUp(self):
         self.e = Exhibitor({'hosts': ['localhost', 'exhibitor'], 'port': 8181, 'scope': 'test',
                             'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 42d01e6b..54b5c674 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -224,6 +224,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch('patroni.postgresql.config.mtime', mock_mtime)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test_check_recovery_conf(self, mock_get_pg_settings):
+        self.p.call_nowait('on_start')
         mock_get_pg_settings.return_value = {
             'primary_conninfo': ['primary_conninfo', 'foo=', None, 'string', 'postmaster', self.p.config._auto_conf],
             'recovery_min_apply_delay': ['recovery_min_apply_delay', '0', 'ms', 'integer', 'sighup', 'foo']
@@ -259,6 +260,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(MockPostmaster, 'create_time', Mock(return_value=1234567), create=True)
     @patch('patroni.postgresql.config.ConfigHandler._get_pg_settings')
     def test__read_recovery_params(self, mock_get_pg_settings):
+        self.p.call_nowait('on_start')
         mock_get_pg_settings.return_value = {'primary_conninfo': ['primary_conninfo', '', None, 'string',
                                                                   'postmaster', self.p.config._postgresql_conf]}
         self.p.config.write_recovery_conf({'standby_mode': 'on', 'primary_conninfo': {'password': 'foo'}})
@@ -725,3 +727,8 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_set_enforce_hot_standby_feedback(self):
         self.p.set_enforce_hot_standby_feedback(True)
+
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=140000))
+    @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=True))
+    def test_handle_parameter_change(self):
+        self.p.handle_parameter_change()
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index ef368826..ed718a21 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -2,12 +2,13 @@ import select
 import six
 import unittest
 
-from kazoo.client import KazooState
+from kazoo.client import KazooClient, KazooState
 from kazoo.exceptions import NoNodeError, NodeExistsError
 from kazoo.handlers.threading import SequentialThreadingHandler
-from kazoo.protocol.states import ZnodeStat
+from kazoo.protocol.states import KeeperState, ZnodeStat
 from mock import Mock, patch
-from patroni.dcs.zookeeper import Leader, PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
+from patroni.dcs.zookeeper import Leader, PatroniKazooClient,\
+        PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
 
 
 class MockKazooClient(Mock):
@@ -128,9 +129,19 @@ class TestPatroniSequentialThreadingHandler(unittest.TestCase):
         self.assertRaises(select.error, self.handler.select)
 
 
+class TestPatroniKazooClient(unittest.TestCase):
+
+    def test__call(self):
+        c = PatroniKazooClient()
+        with patch.object(KazooClient, '_call', Mock()):
+            self.assertIsNotNone(c._call(None, Mock()))
+        c._state = KeeperState.CONNECTING
+        self.assertFalse(c._call(None, Mock()))
+
+
 class TestZooKeeper(unittest.TestCase):
 
-    @patch('patroni.dcs.zookeeper.KazooClient', MockKazooClient)
+    @patch('patroni.dcs.zookeeper.PatroniKazooClient', MockKazooClient)
     def setUp(self):
         self.zk = ZooKeeper({'hosts': ['localhost:2181'], 'scope': 'test',
                              'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10})