From 0e6a2ff3a9e004396d56103a86e278423e392140 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Tue, 5 Dec 2023 08:30:20 +0100
Subject: [PATCH 01/10] Don't let replica restore initialize key when DCS was
 wiped (#2970)

It was happening from the branch where Patroni was supposed to be complain about converting standalone PG cluster to be governed by Patroni and exit.
---
 patroni/ha.py    | 5 ++---
 tests/test_ha.py | 5 +++++
 2 files changed, 7 insertions(+), 3 deletions(-)

diff --git a/patroni/ha.py b/patroni/ha.py
index ab1bc433..0d3e05a4 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -1849,10 +1849,9 @@ class Ha(object):
                             logger.fatal('system ID mismatch, node %s belongs to a different cluster: %s != %s',
                                          self.state_handler.name, self.cluster.initialize, data_sysid)
                             sys.exit(1)
-                elif self.cluster.is_unlocked() and not self.is_paused():
+                elif self.cluster.is_unlocked() and not self.is_paused() and not self.state_handler.cb_called:
                     # "bootstrap", but data directory is not empty
-                    if not self.state_handler.cb_called and self.state_handler.is_running() \
-                            and not self.state_handler.is_primary():
+                    if self.state_handler.is_running() and not self.state_handler.is_primary():
                         self._join_aborted = True
                         logger.error('No initialize key in DCS and PostgreSQL is running as replica, aborting start')
                         logger.error('Please first start Patroni on the node running as primary')
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 40063018..5b1d4562 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -1585,6 +1585,11 @@ class TestHa(PostgresInit):
         self.p.is_primary = false
         self.ha.run_cycle()
         exit_mock.assert_called_once_with(1)
+        self.p.set_role('replica')
+        self.ha.dcs.initialize = Mock()
+        with patch.object(Postgresql, 'cb_called', PropertyMock(return_value=True)):
+            self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
+        self.ha.dcs.initialize.assert_not_called()
 
     @patch.object(Cluster, 'is_unlocked', Mock(return_value=False))
     def test_after_pause(self):

From a4e0a2220dd8dceffc38e23dc44e3ee12a048a8a Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Wed, 6 Dec 2023 15:28:03 +0100
Subject: [PATCH 02/10] Disable SSL for MacOS GH action runners (#2976)

Latest runners release (20231127.1) somehow broke our tests. Connections to postgres somehow failing with strange error:
```
could not accept SSL connection: Socket operation on non-socket
```
---
 features/environment.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/features/environment.py b/features/environment.py
index a0367657..db10c93d 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -1073,6 +1073,8 @@ def before_all(context):
     context.keyfile = os.path.join(context.pctl.output_dir, 'patroni.key')
     context.certfile = os.path.join(context.pctl.output_dir, 'patroni.crt')
     try:
+        if sys.platform == 'darwin' and 'GITHUB_ACTIONS' in os.environ:
+            raise Exception
         with open(os.devnull, 'w') as null:
             ret = subprocess.call(['openssl', 'req', '-nodes', '-new', '-x509', '-subj', '/CN=batman.patroni',
                                    '-addext', 'subjectAltName=IP:127.0.0.1', '-keyout', context.keyfile,

From bbddca6a76bac41ccd5fe142bc198fa7d42fa1ed Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Wed, 6 Dec 2023 15:55:51 +0100
Subject: [PATCH 03/10] Use consistent read when fetching just updated sync key
 (#2974)

Consul doesn't provide any interface to immediately get `ModifyIndex` for the key that we just updated, therefore we have to perform an explicit read operation. By default stale reads are allowed and sometimes we may read stale data. As a result write_sync_state() call was considered as failed. To mitigate the problem we switch to `consistent` reads when that executed after update of the `/sync` key.

Close #2972
---
 patroni/dcs/consul.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index 27cab778..fe66c6d8 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -666,7 +666,7 @@ class Consul(AbstractDCS):
         if ret:  # We have no other choise, only read after write :(
             if not retry.ensure_deadline(0.5):
                 return False
-            _, ret = self.retry(self._client.kv.get, self.sync_path)
+            _, ret = self.retry(self._client.kv.get, self.sync_path, consistency='consistent')
             if ret and (ret.get('Value') or b'').decode('utf-8') == value:
                 return ret['ModifyIndex']
         return False

From efdedc7049527117b35849dc5c01525d7f1133e8 Mon Sep 17 00:00:00 2001
From: Polina Bungina <27892524+hughcapet@users.noreply.github.com>
Date: Wed, 6 Dec 2023 15:57:05 +0100
Subject: [PATCH 04/10] Reload postgres config if a server param was reset
 (#2975)

Fix the case when a parameter value was changed and then reset back to
the initial value without restart - before this fix, the second change
was not reflected in the Postgres config.
This commit also includes the related unit test refactoring.
---
 patroni/postgresql/config.py |   6 ++
 patroni/utils.py             |  21 ++++---
 tests/__init__.py            |  56 +++++++++++------
 tests/test_postgresql.py     | 116 ++++++++++++++++++++++++++++-------
 4 files changed, 150 insertions(+), 49 deletions(-)

diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index 271bbdfe..7dfa8934 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -1098,6 +1098,12 @@ class ConfigHandler(object):
                                     local_connection_address_changed = True
                             else:
                                 logger.info('Changed %s from %s to %s', r[0], r[1], new_value)
+                        elif r[0] in self._server_parameters \
+                                and not compare_values(r[3], r[2], r[1], self._server_parameters[r[0]]):
+                            # Check if any parameter was set back to the current pg_settings value
+                            # We can use pg_settings value here, as it is proved to be equal to new_value
+                            logger.info('Changed %s from %s to %s', r[0], self._server_parameters[r[0]], r[1])
+                            conf_changed = True
                 for param, value in changes.items():
                     if '.' in param:
                         # Check that user-defined-paramters have changed (parameters with period in name)
diff --git a/patroni/utils.py b/patroni/utils.py
index 23f419e5..2b1ded5b 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -400,22 +400,23 @@ def parse_real(value: Any, base_unit: Optional[str] = None) -> Optional[float]:
         return convert_to_base_unit(val, unit, base_unit)
 
 
-def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value: Any) -> bool:
-    """Check if *old_value* and *new_value* are equivalent after parsing them as *vartype*.
+def compare_values(vartype: str, unit: Optional[str], settings_value: Any, config_value: Any) -> bool:
+    """Check if the value from ``pg_settings`` and from Patroni config are equivalent after parsing them as *vartype*.
 
-    :param vartpe: the target type to parse *old_value* and *new_value* before comparing them. Accepts any among of the
-        following (case sensitive):
+    :param vartype: the target type to parse *settings_value* and *config_value* before comparing them.
+        Accepts any among of the following (case sensitive):
 
         * ``bool``: parse values using :func:`parse_bool`; or
         * ``integer``: parse values using :func:`parse_int`; or
         * ``real``: parse values using :func:`parse_real`; or
         * ``enum``: parse values as lowercase strings; or
         * ``string``: parse values as strings. This one is used by default if no valid value is passed as *vartype*.
-    :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real` for *new_value*.
-    :param old_value: value to be compared with *new_value*.
-    :param new_value: value to be compared with *old_value*.
+    :param unit: base unit to be used as argument when calling :func:`parse_int` or :func:`parse_real`
+        for *config_value*.
+    :param settings_value: value to be compared with *config_value*.
+    :param config_value: value to be compared with *settings_value*.
 
-    :returns: ``True`` if *old_value* is equivalent to *new_value* when both are parsed as *vartype*.
+    :returns: ``True`` if *settings_value* is equivalent to *config_value* when both are parsed as *vartype*.
 
     :Example:
 
@@ -455,8 +456,8 @@ def compare_values(vartype: str, unit: Optional[str], old_value: Any, new_value:
     }
 
     converter = converters.get(vartype) or converters['string']
-    old_converted = converter(old_value, None)
-    new_converted = converter(new_value, unit)
+    old_converted = converter(settings_value, None)
+    new_converted = converter(config_value, unit)
 
     return old_converted is not None and new_converted is not None and old_converted == new_converted
 
diff --git a/tests/__init__.py b/tests/__init__.py
index bd70ba3d..b013e4e1 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -25,8 +25,41 @@ mock_available_gucs = PropertyMock(return_value={
     'max_wal_senders', 'max_worker_processes', 'port', 'search_path', 'shared_preload_libraries',
     'stats_temp_directory', 'synchronous_standby_names', 'track_commit_timestamp', 'unix_socket_directories',
     'vacuum_cost_delay', 'vacuum_cost_limit', 'wal_keep_size', 'wal_level', 'wal_log_hints', 'zero_damaged_pages',
+    'autovacuum', 'wal_segment_size', 'wal_block_size', 'shared_buffers', 'wal_buffers',
 })
 
+GET_PG_SETTINGS_RESULT = [
+    ('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
+    ('wal_block_size', '8192', None, 'integer', 'internal'),
+    ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
+    ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
+    ('max_connections', '100', None, 'integer', 'postmaster'),
+    ('max_prepared_transactions', '200', None, 'integer', 'postmaster'),
+    ('max_worker_processes', '8', None, 'integer', 'postmaster'),
+    ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
+    ('max_wal_senders', '5', None, 'integer', 'postmaster'),
+    ('search_path', 'public', None, 'string', 'user'),
+    ('port', '5432', None, 'integer', 'postmaster'),
+    ('listen_addresses', '127.0.0.2, 127.0.0.3', None, 'string', 'postmaster'),
+    ('autovacuum', 'on', None, 'bool', 'sighup'),
+    ('unix_socket_directories', '/tmp', None, 'string', 'postmaster'),
+    ('shared_preload_libraries', 'citus', None, 'string', 'postmaster'),
+    ('wal_keep_size', '128', 'MB', 'integer', 'sighup'),
+    ('cluster_name', 'batman', None, 'string', 'postmaster'),
+    ('vacuum_cost_delay', '200', 'ms', 'real', 'user'),
+    ('vacuum_cost_limit', '-1', None, 'integer', 'user'),
+    ('max_stack_depth', '2048', 'kB', 'integer', 'superuser'),
+    ('constraint_exclusion', '', None, 'enum', 'user'),
+    ('force_parallel_mode', '1', None, 'enum', 'user'),
+    ('zero_damaged_pages', 'off', None, 'bool', 'superuser'),
+    ('stats_temp_directory', '/tmp', None, 'string', 'sighup'),
+    ('track_commit_timestamp', 'off', None, 'bool', 'postmaster'),
+    ('wal_log_hints', 'on', None, 'bool', 'superuser'),
+    ('hot_standby', 'on', None, 'bool', 'superuser'),
+    ('max_replication_slots', '5', None, 'integer', 'superuser'),
+    ('wal_level', 'logical', None, 'enum', 'superuser'),
+]
+
 
 class MockResponse(object):
 
@@ -133,22 +166,9 @@ class MockCursor(object):
                             ('archive_command', 'my archive command'),
                             ('cluster_name', 'my_cluster')]
         elif sql.startswith('SELECT name, setting'):
-            self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
-                            ('wal_block_size', '8192', None, 'integer', 'internal'),
-                            ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
-                            ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
-                            ('max_connections', '100', None, 'integer', 'postmaster'),
-                            ('max_prepared_transactions', '0', None, 'integer', 'postmaster'),
-                            ('max_worker_processes', '8', None, 'integer', 'postmaster'),
-                            ('max_locks_per_transaction', '64', None, 'integer', 'postmaster'),
-                            ('max_wal_senders', '5', None, 'integer', 'postmaster'),
-                            ('search_path', 'public', None, 'string', 'user'),
-                            ('port', '5433', None, 'integer', 'postmaster'),
-                            ('listen_addresses', '*', None, 'string', 'postmaster'),
-                            ('autovacuum', 'on', None, 'bool', 'sighup'),
-                            ('unix_socket_directories', '/tmp', None, 'string', 'postmaster')]
+            self.results = GET_PG_SETTINGS_RESULT
         elif sql.startswith('SELECT COUNT(*) FROM pg_catalog.pg_settings'):
-            self.results = [(1,)]
+            self.results = [(0,)]
         elif sql.startswith('IDENTIFY_SYSTEM'):
             self.results = [('1', 3, '0/402EEC0', '')]
         elif sql.startswith('TIMELINE_HISTORY '):
@@ -218,11 +238,11 @@ class PostgresInit(unittest.TestCase):
     _PARAMETERS = {'wal_level': 'hot_standby', 'max_replication_slots': 5, 'f.oo': 'bar',
                    'search_path': 'public', 'hot_standby': 'on', 'max_wal_senders': 5,
                    'wal_keep_segments': 8, 'wal_log_hints': 'on', 'max_locks_per_transaction': 64,
-                   'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 0,
+                   'max_worker_processes': 8, 'max_connections': 100, 'max_prepared_transactions': 200,
                    'track_commit_timestamp': 'off', 'unix_socket_directories': '/tmp',
-                   'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': '',
+                   'trigger_file': 'bla', 'stats_temp_directory': '/tmp', 'zero_damaged_pages': 'off',
                    'force_parallel_mode': '1', 'constraint_exclusion': '',
-                   'max_stack_depth': 'Z', 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}
+                   'max_stack_depth': 2048, 'vacuum_cost_limit': -1, 'vacuum_cost_delay': 200}
 
     @patch('patroni.psycopg._connect', psycopg_connect)
     @patch('patroni.postgresql.CallbackExecutor', Mock())
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 31454479..8a5c491b 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -5,6 +5,7 @@ import re
 import subprocess
 import time
 
+from copy import deepcopy
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 
 import patroni.psycopg as psycopg
@@ -25,7 +26,8 @@ from patroni.postgresql.validator import (ValidatorFactoryNoType, ValidatorFacto
 from patroni.utils import RetryFailedError
 from threading import Thread, current_thread
 
-from . import BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs
+from . import (BaseTestPostgresql, MockCursor, MockPostmaster, psycopg_connect, mock_available_gucs,
+               GET_PG_SETTINGS_RESULT)
 
 
 mtime_ret = {}
@@ -559,31 +561,103 @@ class TestPostgresql(BaseTestPostgresql):
 
     @patch('time.sleep', Mock())
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
-    def test_reload_config(self):
-        parameters = self._PARAMETERS.copy()
-        parameters.pop('f.oo')
-        parameters['wal_buffers'] = '512'
-        config = {'pg_hba': [''], 'pg_ident': [''], 'use_unix_socket': True, 'use_unix_socket_repl': True,
-                  'authentication': {},
-                  'retry_timeout': 10, 'listen': '*', 'krbsrvname': 'postgres', 'parameters': parameters}
+    @patch('patroni.postgresql.config.logger.info')
+    @patch('patroni.postgresql.config.logger.warning')
+    def test_reload_config(self, mock_warning, mock_info):
+        config = deepcopy(self.p.config._config)
+
+        # Nothing changed
         self.p.reload_config(config)
-        parameters['b.ar'] = 'bar'
-        with patch.object(MockCursor, 'fetchall',
-                          Mock(side_effect=[[('wal_block_size', '8191', None, 'integer', 'internal'),
-                                             ('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
-                                             ('shared_buffers', '16384', '8kB', 'integer', 'postmaster'),
-                                             ('wal_buffers', '-1', '8kB', 'integer', 'postmaster'),
-                                             ('port', '5433', None, 'integer', 'postmaster')], Exception])):
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        mock_warning.assert_not_called()
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Handle wal_buffers
+        self.p.config._config['parameters']['wal_buffers'] = '512'
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('No PostgreSQL configuration items changed, nothing to reload.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+        config = deepcopy(self.p.config._config)
+
+        # hba/ident_changed
+        config['pg_hba'] = ['']
+        config['pg_ident'] = ['']
+        self.p.reload_config(config)
+        mock_info.assert_called_once_with('Reloading PostgreSQL configuration.')
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Postmaster parameter change (pending_restart)
+        init_max_worker_processes = config['parameters']['max_worker_processes']
+        config['parameters']['max_worker_processes'] *= 2
+        with patch('patroni.postgresql.Postgresql._query', Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)]])):
             self.p.reload_config(config)
-        parameters['autovacuum'] = 'on'
+            self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s (restart might be required)',
+                                                              'max_worker_processes', str(init_max_worker_processes),
+                                                              config['parameters']['max_worker_processes']))
+            self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+            self.assertEqual(self.p.pending_restart, True)
+
+        mock_info.reset_mock()
+
+        # Reset to the initial value without restart
+        config['parameters']['max_worker_processes'] = init_max_worker_processes
         self.p.reload_config(config)
-        parameters['autovacuum'] = 'off'
-        parameters.pop('search_path')
-        config['listen'] = '*:5433'
+        self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'max_worker_processes',
+                                                          init_max_worker_processes * 2,
+                                                          str(config['parameters']['max_worker_processes'])))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # User-defined parameter changed (removed)
+        config['parameters'].pop('f.oo')
         self.p.reload_config(config)
-        parameters['unix_socket_directories'] = '.'
+        self.assertEqual(mock_info.call_args_list[0][0], ('Changed %s from %s to %s', 'f.oo', 'bar', None))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        mock_info.reset_mock()
+
+        # Non-postmaster parameter change
+        config['parameters']['autovacuum'] = 'off'
         self.p.reload_config(config)
-        self.p.config.resolve_connection_addresses()
+        self.assertEqual(mock_info.call_args_list[0][0], ("Changed %s from %s to %s", 'autovacuum', 'on', 'off'))
+        self.assertEqual(mock_info.call_args_list[1][0], ('Reloading PostgreSQL configuration.',))
+        self.assertEqual(self.p.pending_restart, False)
+
+        config['parameters']['autovacuum'] = 'on'
+        mock_info.reset_mock()
+
+        # Remove invalid parameter
+        config['parameters']['invalid'] = 'value'
+        self.p.reload_config(config)
+        self.assertEqual(mock_warning.call_args_list[0][0],
+                         ('Removing invalid parameter `%s` from postgresql.parameters', 'invalid'))
+        config['parameters'].pop('invalid')
+
+        mock_warning.reset_mock()
+        mock_info.reset_mock()
+
+        # Non-empty result (outside changes) and exception while querying pending_restart parameters
+        with patch('patroni.postgresql.Postgresql._query',
+                   Mock(side_effect=[GET_PG_SETTINGS_RESULT, [(1,)], GET_PG_SETTINGS_RESULT, Exception])):
+            self.p.reload_config(config, True)
+            self.assertEqual(mock_info.call_args_list[0][0], ('Reloading PostgreSQL configuration.',))
+            self.assertEqual(self.p.pending_restart, True)
+
+            # Invalid values, just to increase silly coverage in postgresql.validator.
+            # One day we will have proper tests there.
+            config['parameters']['autovacuum'] = 'of'  # Bool.transform()
+            config['parameters']['vacuum_cost_limit'] = 'smth'  # Number.transform()
+            self.p.reload_config(config, True)
+            self.assertEqual(mock_warning.call_args_list[-1][0][0], 'Exception %r when running query')
 
     def test_resolve_connection_addresses(self):
         self.p.config._config['use_unix_socket'] = self.p.config._config['use_unix_socket_repl'] = True

From f0719d148c54dfc6a73ef848025e9d162d9c2d39 Mon Sep 17 00:00:00 2001
From: Polina Bungina <27892524+hughcapet@users.noreply.github.com>
Date: Wed, 13 Dec 2023 08:40:47 +0100
Subject: [PATCH 05/10] Actually allow failover to an async candidate in sync
 mode (#2980)

---
 patroni/ctl.py    |  2 +-
 tests/test_ctl.py | 10 ++++++----
 2 files changed, 7 insertions(+), 5 deletions(-)

diff --git a/patroni/ctl.py b/patroni/ctl.py
index 3e981b45..3a0d2a17 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -1272,7 +1272,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
             config.is_synchronous_mode,
             not cluster.sync.is_empty,
             not cluster.sync.matches(candidate, True))):
-        if click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}'):
+        if not click.confirm(f'Are you sure you want to failover to the asynchronous node {candidate}?'):
             raise PatroniCtlException('Aborting ' + action)
 
     scheduled_at_str = None
diff --git a/tests/test_ctl.py b/tests/test_ctl.py
index a174b03d..badbeca1 100644
--- a/tests/test_ctl.py
+++ b/tests/test_ctl.py
@@ -226,18 +226,20 @@ class TestCtl(unittest.TestCase):
             self.assertIn('Supplying a leader name using this command is deprecated', result.output)
             failover_func_mock.assert_called_once_with('switchover', 'dummy', None, 'leader', None, False)
 
-        # Failover to an async member in sync mode (confirm)
         cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
         cluster.members.append(Member(0, 'async', 28, {'api_url': 'http://127.0.0.1:8012/patroni'}))
         cluster.config.data['synchronous_mode'] = True
         with patch('patroni.dcs.AbstractDCS.get_cluster', Mock(return_value=cluster)):
+            # Failover to an async member in sync mode (confirm)
             result = self.runner.invoke(ctl,
                                         ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='y\ny')
             self.assertIn('Are you sure you want to failover to the asynchronous node async', result.output)
+            self.assertEqual(result.exit_code, 0)
 
-        # Failover to an async member in sync mode (abort)
-        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N')
-        self.assertEqual(result.exit_code, 1)
+            # Failover to an async member in sync mode (abort)
+            result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0', '--candidate', 'async'], input='N')
+            self.assertEqual(result.exit_code, 1)
+            self.assertIn('Aborting failover', result.output)
 
     @patch('patroni.dynamic_loader.iter_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd']))
     def test_get_dcs(self):

From c1ee99d81da8b35e370350a433c5f5c4889baff5 Mon Sep 17 00:00:00 2001
From: Polina Bungina <27892524+hughcapet@users.noreply.github.com>
Date: Mon, 18 Dec 2023 10:44:05 +0100
Subject: [PATCH 06/10] Update PG version in a couple of places (#2986)

* All dockerfiles to use PG16 by default
* PGVERSION env in the test pipelines to 16.1-1 by default
* 11->14 in the dcs-pg mapping for test pipelines
* Code comments fixes
---
 .github/workflows/install_deps.py | 2 +-
 .github/workflows/mapping.py      | 2 +-
 .github/workflows/run_tests.py    | 2 +-
 .github/workflows/tests.yaml      | 2 +-
 Dockerfile                        | 2 +-
 Dockerfile.citus                  | 4 ++--
 kubernetes/Dockerfile             | 2 +-
 kubernetes/Dockerfile.citus       | 8 ++++----
 patroni/postgresql/postmaster.py  | 2 +-
 9 files changed, 13 insertions(+), 13 deletions(-)

diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index 6480f66a..b089bde0 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -110,7 +110,7 @@ def install_etcd():
 
 
 def install_postgres():
-    version = os.environ.get('PGVERSION', '15.1-1')
+    version = os.environ.get('PGVERSION', '16.1-1')
     platform = {'darwin': 'osx', 'win32': 'windows-x64', 'cygwin': 'windows-x64'}[sys.platform]
     if platform == 'osx':
         return subprocess.call(['brew', 'install', 'expect', 'postgresql@{0}'.format(version.split('.')[0])])
diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py
index f75efec4..279438b0 100644
--- a/.github/workflows/mapping.py
+++ b/.github/workflows/mapping.py
@@ -1 +1 @@
-versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '15'}
+versions = {'etcd': '9.6', 'etcd3': '16', 'consul': '13', 'exhibitor': '12', 'raft': '14', 'kubernetes': '15'}
diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py
index 9a078e4f..cece186f 100644
--- a/.github/workflows/run_tests.py
+++ b/.github/workflows/run_tests.py
@@ -30,7 +30,7 @@ def main():
         unbuffer = ['timeout', '900', 'unbuffer']
     else:
         if sys.platform == 'darwin':
-            version = os.environ.get('PGVERSION', '15.1-1')
+            version = os.environ.get('PGVERSION', '16.1-1')
             path = '/usr/local/opt/postgresql@{0}/bin:.'.format(version.split('.')[0])
             unbuffer = ['unbuffer']
         else:
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index aafe6b56..f1f55e2a 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -85,7 +85,7 @@ jobs:
     env:
       DCS: ${{ matrix.dcs }}
       ETCDVERSION: 3.4.23
-      PGVERSION: 15.1-1  # for windows and macos
+      PGVERSION: 16.1-1  # for windows and macos
     strategy:
       fail-fast: false
       matrix:
diff --git a/Dockerfile b/Dockerfile
index b74cdf24..d4fddfea 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,6 +1,6 @@
 ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine
 ## It has all the necessary components to play/debug with a single node appliance, running etcd
-ARG PG_MAJOR=15
+ARG PG_MAJOR=16
 ARG COMPRESS=false
 ARG PGHOME=/home/postgres
 ARG PGDATA=$PGHOME/data
diff --git a/Dockerfile.citus b/Dockerfile.citus
index 5f0164b4..6f02215b 100644
--- a/Dockerfile.citus
+++ b/Dockerfile.citus
@@ -1,6 +1,6 @@
 ## This Dockerfile is meant to aid in the building and debugging patroni whilst developing on your local machine
 ## It has all the necessary components to play/debug with a single node appliance, running etcd
-ARG PG_MAJOR=15
+ARG PG_MAJOR=16
 ARG COMPRESS=false
 ARG PGHOME=/home/postgres
 ARG PGDATA=$PGHOME/data
@@ -40,7 +40,7 @@ RUN set -ex \
         echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
         && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
         && apt-get update -y \
-        && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \
+        && apt-get -y install postgresql-$PG_MAJOR-citus-12.1; \
     fi \
 \
     # Cleanup all locales but en_US.UTF-8
diff --git a/kubernetes/Dockerfile b/kubernetes/Dockerfile
index 29a683bd..e41bf1cd 100644
--- a/kubernetes/Dockerfile
+++ b/kubernetes/Dockerfile
@@ -1,4 +1,4 @@
-FROM postgres:15
+FROM postgres:16
 LABEL maintainer="Alexander Kukushkin <akukushkin@microsoft.com>"
 
 RUN export DEBIAN_FRONTEND=noninteractive \
diff --git a/kubernetes/Dockerfile.citus b/kubernetes/Dockerfile.citus
index f9564521..7af9e5ae 100644
--- a/kubernetes/Dockerfile.citus
+++ b/kubernetes/Dockerfile.citus
@@ -1,4 +1,4 @@
-FROM postgres:15
+FROM postgres:16
 LABEL maintainer="Alexander Kukushkin <akukushkin@microsoft.com>"
 
 RUN export DEBIAN_FRONTEND=noninteractive \
@@ -11,7 +11,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \
     ## Make sure we have a en_US.UTF-8 locale available
     && localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8 \
     && if [ $(dpkg --print-architecture) = 'arm64' ]; then \
-        apt-get install -y postgresql-server-dev-15 \
+        apt-get install -y postgresql-server-dev-16 \
                            gcc make autoconf \
                            libc6-dev flex libcurl4-gnutls-dev \
                            libicu-dev libkrb5-dev liblz4-dev \
@@ -24,7 +24,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \
         echo "deb [signed-by=/etc/apt/trusted.gpg.d/citusdata_community.gpg] https://packagecloud.io/citusdata/community/debian/ $(lsb_release -cs) main" > /etc/apt/sources.list.d/citusdata_community.list \
         && curl -sL https://packagecloud.io/citusdata/community/gpgkey | gpg --dearmor > /etc/apt/trusted.gpg.d/citusdata_community.gpg \
         && apt-get update -y \
-        && apt-get -y install postgresql-15-citus-12.0; \
+        && apt-get -y install postgresql-16-citus-12.1; \
     fi \
     && pip3 install --break-system-packages setuptools \
     && pip3 install --break-system-packages 'git+https://github.com/zalando/patroni.git#egg=patroni[kubernetes]' \
@@ -38,7 +38,7 @@ RUN export DEBIAN_FRONTEND=noninteractive \
     && chmod 664 /etc/passwd \
     # Clean up
     && apt-get remove -y git python3-pip python3-wheel \
-               postgresql-server-dev-15 gcc make autoconf \
+               postgresql-server-dev-16 gcc make autoconf \
                libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \
                libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \
     && apt-get autoremove -y \
diff --git a/patroni/postgresql/postmaster.py b/patroni/postgresql/postmaster.py
index 4505e7f7..97eb10e4 100644
--- a/patroni/postgresql/postmaster.py
+++ b/patroni/postgresql/postmaster.py
@@ -176,7 +176,7 @@ class PostmasterProcess(psutil.Process):
             return not self.is_running()
 
     def wait_for_user_backends_to_close(self, stop_timeout: Optional[float]) -> None:
-        # These regexps are cross checked against versions PostgreSQL 9.1 .. 15
+        # These regexps are cross checked against versions PostgreSQL 9.1 .. 16
         aux_proc_re = re.compile("(?:postgres:)( .*:)? (?:(?:archiver|startup|autovacuum launcher|autovacuum worker|"
                                  "checkpointer|logger|stats collector|wal receiver|wal writer|writer)(?: process  )?|"
                                  "walreceiver|wal sender process|walsender|walwriter|background writer|"

From 206ee91b07eed4c405cfabb724105df974038a99 Mon Sep 17 00:00:00 2001
From: Polina Bungina <27892524+hughcapet@users.noreply.github.com>
Date: Wed, 20 Dec 2023 09:54:04 +0100
Subject: [PATCH 07/10] Exclude leader from failover candidates in ctl (#2983)

Exclude actual leader (not the passed leader argument) from the
candidates list in the `patronictl failover` prompt.
Abort `patronictl failover` execution if candidate specified is
the same as the current cluster leader
---
 patroni/ctl.py    | 36 +++++++++++++++++++-----------------
 tests/test_ctl.py |  8 +++++++-
 2 files changed, 26 insertions(+), 18 deletions(-)

diff --git a/patroni/ctl.py b/patroni/ctl.py
index 3a0d2a17..de1fe892 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -1185,7 +1185,7 @@ def reinit(cluster_name: str, group: Optional[int], member_names: List[str], for
 
 
 def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[int],
-                               leader: Optional[str], candidate: Optional[str],
+                               switchover_leader: Optional[str], candidate: Optional[str],
                                force: bool, scheduled: Optional[str] = None) -> None:
     """Perform a failover or a switchover operation in the cluster.
 
@@ -1199,7 +1199,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
     :param cluster_name: name of the Patroni cluster.
     :param group: filter Citus group within we should perform a failover or switchover. If ``None``, user will be
         prompted for filling it -- unless *force* is ``True``, in which case an exception is raised.
-    :param leader: name of the current leader member.
+    :param switchover_leader: name of the leader member passed as switchover option.
     :param candidate: name of a standby member to be promoted. Nodes that are tagged with ``nofailover`` cannot be used.
     :param force: perform the failover or switchover without asking for confirmations.
     :param scheduled: timestamp when the switchover should be scheduled to occur. If ``now`` perform immediately.
@@ -1208,10 +1208,11 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
         :class:`PatroniCtlException`: if:
             * Patroni is running on a Citus cluster, but no *group* was specified; or
             * a switchover was requested by the cluster has no leader; or
-            * *leader* does not match the current leader of the cluster; or
+            * *switchover_leader* does not match the current leader of the cluster; or
             * cluster has no candidates available for the operation; or
             * no *candidate* is given for a failover operation; or
-            * *leader* and *candidate* are the same; or
+            * current leader and *candidate* are the same; or
+            * *candidate* is tagged as nofailover; or
             * *candidate* is not a member of the cluster; or
             * trying to schedule a switchover in a cluster that is in maintenance mode; or
             * user aborts the operation.
@@ -1231,23 +1232,24 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
 
     config = global_config.from_cluster(cluster)
 
+    cluster_leader = cluster.leader and cluster.leader.name
     # leader has to be be defined for switchover only
     if action == 'switchover':
-        if cluster.leader is None or not cluster.leader.name:
+        if not cluster_leader:
             raise PatroniCtlException('This cluster has no leader')
 
-        if leader is None:
+        if switchover_leader is None:
             if force:
-                leader = cluster.leader.name
+                switchover_leader = cluster_leader
             else:
                 prompt = 'Standby Leader' if config.is_standby_cluster else 'Primary'
-                leader = click.prompt(prompt, type=str, default=(cluster.leader and cluster.leader.name))
+                switchover_leader = click.prompt(prompt, type=str, default=cluster_leader)
 
-        if cluster.leader.name != leader:
-            raise PatroniCtlException(f'Member {leader} is not the leader of cluster {cluster_name}')
+        if cluster_leader != switchover_leader:
+            raise PatroniCtlException(f'Member {switchover_leader} is not the leader of cluster {cluster_name}')
 
     # excluding members with nofailover tag
-    candidate_names = [str(m.name) for m in cluster.members if m.name != leader and not m.nofailover]
+    candidate_names = [str(m.name) for m in cluster.members if m.name != cluster_leader and not m.nofailover]
     # We sort the names for consistent output to the client
     candidate_names.sort()
 
@@ -1260,10 +1262,10 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
     if action == 'failover' and not candidate:
         raise PatroniCtlException('Failover could be performed only to a specific candidate')
 
-    if candidate == leader:
-        raise PatroniCtlException(action.title() + ' target and source are the same.')
-
     if candidate and candidate not in candidate_names:
+        if candidate == cluster_leader:
+            raise PatroniCtlException(
+                f'Member {candidate} is already the leader of cluster {cluster_name}')
         raise PatroniCtlException(
             f'Member {candidate} does not exist in cluster {cluster_name} or is tagged as nofailover')
 
@@ -1292,7 +1294,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
 
     failover_value = {'candidate': candidate}
     if action == 'switchover':
-        failover_value['leader'] = leader
+        failover_value['leader'] = switchover_leader
     if scheduled_at_str:
         failover_value['scheduled_at'] = scheduled_at_str
 
@@ -1300,7 +1302,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
 
     # By now we have established that the leader exists and the candidate exists
     if not force:
-        demote_msg = f', demoting current leader {cluster.leader.name}' if cluster.leader else ''
+        demote_msg = f', demoting current leader {cluster_leader}' if cluster_leader else ''
         if scheduled_at_str:
             # only switchover can be scheduled
             if not click.confirm(f'Are you sure you want to schedule switchover of cluster '
@@ -1334,7 +1336,7 @@ def _do_failover_or_switchover(action: str, cluster_name: str, group: Optional[i
         logging.exception(r)
         logging.warning('Failing over to DCS')
         click.echo('{0} Could not {1} using Patroni api, falling back to DCS'.format(timestamp(), action))
-        dcs.manual_failover(leader, candidate, scheduled_at=scheduled_at)
+        dcs.manual_failover(switchover_leader, candidate, scheduled_at=scheduled_at)
 
     output_members(cluster, cluster_name, group=group)
 
diff --git a/tests/test_ctl.py b/tests/test_ctl.py
index badbeca1..f9ee62ce 100644
--- a/tests/test_ctl.py
+++ b/tests/test_ctl.py
@@ -157,7 +157,8 @@ class TestCtl(unittest.TestCase):
         # Target and source are equal
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nleader\n\ny')
         self.assertEqual(result.exit_code, 1)
-        self.assertIn('Switchover target and source are the same', result.output)
+        self.assertIn("Candidate ['other']", result.output)
+        self.assertIn('Member leader is already the leader of cluster dummy', result.output)
 
         # Candidate is not a member of the cluster
         result = self.runner.invoke(ctl, ['switchover', 'dummy', '--group', '0'], input='leader\nReality\n\ny')
@@ -220,6 +221,11 @@ class TestCtl(unittest.TestCase):
         result = self.runner.invoke(ctl, ['failover', 'dummy'], input='0\n')
         self.assertIn('Failover could be performed only to a specific candidate', result.output)
 
+        # Candidate is the same as the leader
+        result = self.runner.invoke(ctl, ['failover', 'dummy', '--group', '0'], input='leader\n')
+        self.assertIn("Candidate ['other']", result.output)
+        self.assertIn('Member leader is already the leader of cluster dummy', result.output)
+
         # Temp test to check a fallback to switchover if leader is specified
         with patch('patroni.ctl._do_failover_or_switchover') as failover_func_mock:
             result = self.runner.invoke(ctl, ['failover', '--leader', 'leader', 'dummy'], input='0\n')

From 5c3e1a693e2217e990ae42cd27c1d9795df38d68 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Wed, 20 Dec 2023 10:49:33 +0100
Subject: [PATCH 08/10] Implement validation of the `log` section (#2989)

Somehow it was always forgotten.
---
 patroni/validator.py    | 12 ++++++++++++
 tests/test_validator.py | 14 ++++++++++++++
 2 files changed, 26 insertions(+)

diff --git a/patroni/validator.py b/patroni/validator.py
index d0b168be..bd69cc0e 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -937,6 +937,18 @@ validate_etcd = {
 schema = Schema({
     "name": str,
     "scope": str,
+    Optional("log"): {
+        Optional("level"): EnumValidator(('DEBUG', 'INFO', 'WARN', 'WARNING', 'ERROR', 'FATAL', 'CRITICAL'),
+                                         case_sensitive=True, raise_assert=True),
+        Optional("traceback_level"): EnumValidator(('DEBUG', 'ERROR'), raise_assert=True),
+        Optional("format"): str,
+        Optional("dateformat"): str,
+        Optional("max_queue_size"): int,
+        Optional("dir"): str,
+        Optional("file_num"): int,
+        Optional("file_size"): int,
+        Optional("loggers"): dict
+    },
     Optional("ctl"): {
         Optional("insecure"): bool,
         Optional("cacert"): str,
diff --git a/tests/test_validator.py b/tests/test_validator.py
index 1f647dbe..d5e82992 100644
--- a/tests/test_validator.py
+++ b/tests/test_validator.py
@@ -13,6 +13,20 @@ available_dcs = [m.split(".")[-1] for m in dcs_modules()]
 config = {
     "name": "string",
     "scope": "string",
+    "log": {
+        "level": "DEBUG",
+        "traceback_level": "DEBUG",
+        "format": "%(asctime)s %(levelname)s: %(message)s",
+        "dateformat": "%Y-%m-%d %H:%M:%S",
+        "max_queue_size": 100,
+        "dir": "/tmp",
+        "file_num": 10,
+        "file_size": 1000000,
+        "loggers": {
+            "patroni.postmaster": "WARNING",
+            "urllib3": "DEBUG"
+        }
+    },
     "restapi": {
         "listen": "127.0.0.2:800",
         "connect_address": "127.0.0.2:800",

From bcfd8438a50a1f36b65be76ee2d17d1da3b903e3 Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Thu, 21 Dec 2023 08:58:26 +0100
Subject: [PATCH 09/10] Abstract CitusHandler and decouple it from
 configuration (#2950)

the main issue was that the configuration for Citus handler and for DCS existed in two places, while ideally AbstractDCS should not know many details about what kind of MPP is in use.

To solve the problem we first dynamically create an object implementing AbstractMPP interfaces, which is a configuration for DCS. Later this object is used to instantiate the class implementing AbstractMPPHandler interface.

This is just a starting point, which does some heavy lifting. As a next steps all kind of variables named after Citus in files different from patroni/postgres/mpp/citus.py should be renamed.

In other words this commit takes over the most complex part of #2940, which was never implemented.

Co-authored-by: zhjwpku <zhjwpku@gmail.com>
---
 patroni/__main__.py                   |   2 +-
 patroni/api.py                        |   4 +-
 patroni/ctl.py                        |   8 +-
 patroni/dcs/__init__.py               |  33 +--
 patroni/dcs/consul.py                 |   9 +-
 patroni/dcs/etcd.py                   |  13 +-
 patroni/dcs/etcd3.py                  |  10 +-
 patroni/dcs/exhibitor.py              |   5 +-
 patroni/dcs/kubernetes.py             |  20 +-
 patroni/dcs/raft.py                   |  10 +-
 patroni/dcs/zookeeper.py              |  10 +-
 patroni/ha.py                         |   4 +-
 patroni/postgresql/__init__.py        |   6 +-
 patroni/postgresql/mpp/__init__.py    | 296 ++++++++++++++++++++++++++
 patroni/postgresql/{ => mpp}/citus.py | 103 ++++++---
 patroni/postgresql/slots.py           |   2 +-
 tests/__init__.py                     |  36 ++--
 tests/test_citus.py                   |  34 ++-
 tests/test_consul.py                  |  22 +-
 tests/test_ctl.py                     |  11 +-
 tests/test_etcd.py                    |  11 +-
 tests/test_etcd3.py                   |  10 +-
 tests/test_exhibitor.py               |   6 +-
 tests/test_ha.py                      |  22 +-
 tests/test_kubernetes.py              |  17 +-
 tests/test_mpp.py                     |  52 +++++
 tests/test_raft.py                    |  21 +-
 tests/test_zookeeper.py               |  10 +-
 28 files changed, 602 insertions(+), 185 deletions(-)
 create mode 100644 patroni/postgresql/mpp/__init__.py
 rename patroni/postgresql/{ => mpp}/citus.py (84%)
 create mode 100644 tests/test_mpp.py

diff --git a/patroni/__main__.py b/patroni/__main__.py
index 229ccfb9..2c253da4 100644
--- a/patroni/__main__.py
+++ b/patroni/__main__.py
@@ -68,7 +68,7 @@ class Patroni(AbstractPatroniDaemon, Tags):
         self.watchdog = Watchdog(self.config)
         self.load_dynamic_configuration()
 
-        self.postgresql = Postgresql(self.config['postgresql'])
+        self.postgresql = Postgresql(self.config['postgresql'], self.dcs.mpp)
         self.api = RestApiServer(self, self.config['restapi'])
         self.ha = Ha(self)
 
diff --git a/patroni/api.py b/patroni/api.py
index 5761d359..5c9d6603 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -1153,8 +1153,8 @@ class RestApiHandler(BaseHTTPRequestHandler):
     def do_POST_citus(self) -> None:
         """Handle a ``POST`` request to ``/citus`` path.
 
-        Call :func:`~patroni.postgresql.CitusHandler.handle_event` to handle the request, then write a response with
-        HTTP status code ``200``.
+        Call :func:`~patroni.postgresql.mpp.AbstractMPPHandler.handle_event` to handle the request,
+        then write a response with HTTP status code ``200``.
 
         .. note::
             If unable to parse the request body, then the request is silently discarded.
diff --git a/patroni/ctl.py b/patroni/ctl.py
index de1fe892..6a5fe0e6 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -51,6 +51,7 @@ from .config import Config
 from .dcs import get_dcs as _get_dcs, AbstractDCS, Cluster, Member
 from .exceptions import PatroniException
 from .postgresql.misc import postgres_version_to_int
+from .postgresql.mpp import get_mpp
 from .utils import cluster_as_json, patch_config, polling_loop
 from .request import PatroniRequest
 from .version import __version__
@@ -313,7 +314,7 @@ def ctl(ctx: click.Context, config_file: str, dcs_url: Optional[str], insecure:
     config = load_config(config_file, dcs_url)
     # backward compatibility for configuration file where ctl section is not defined
     config.setdefault('ctl', {})['insecure'] = config.get('ctl', {}).get('insecure') or insecure
-    ctx.obj = {'__config': config}
+    ctx.obj = {'__config': config, '__mpp': get_mpp(config)}
 
 
 def is_citus_cluster() -> bool:
@@ -321,7 +322,7 @@ def is_citus_cluster() -> bool:
 
     :returns: ``True`` if configuration has ``citus`` section, otherwise ``False``.
     """
-    return bool(_get_configuration().get('citus'))
+    return click.get_current_context().obj['__mpp'].is_enabled()
 
 
 def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS:
@@ -340,12 +341,13 @@ def get_dcs(scope: str, group: Optional[int]) -> AbstractDCS:
     config = _get_configuration()
     config.update({'scope': scope, 'patronictl': True})
     if group is not None:
-        config['citus'] = {'group': group}
+        config['citus'] = {'group': group, 'database': 'postgres'}
     config.setdefault('name', scope)
     try:
         dcs = _get_dcs(config)
         if is_citus_cluster() and group is None:
             dcs.is_citus_coordinator = lambda: True
+        click.get_current_context().obj['__mpp'] = dcs.mpp
         return dcs
     except PatroniException as e:
         raise PatroniCtlException(str(e))
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index 28c3734f..a210615b 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -25,10 +25,9 @@ from ..utils import parse_int
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
     from ..postgresql import Postgresql
+    from ..postgresql.mpp import AbstractMPP
 
 SLOT_ADVANCE_AVAILABLE_VERSION = 110000
-CITUS_COORDINATOR_GROUP_ID = 0
-citus_group_re = re.compile('^(0|[1-9][0-9]*)$')
 slot_name_re = re.compile('^[a-z0-9_]{1,63}$')
 logger = logging.getLogger(__name__)
 
@@ -130,10 +129,9 @@ def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
             p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait',
                                    'patronictl', 'ttl', 'retry_timeout')
             if p in config})
-        # From citus section we only need "group" parameter, but will propagate everything just in case.
-        if isinstance(config.get('citus'), dict):
-            config[name].update(config['citus'])
-        return dcs_class(config[name])
+
+        from patroni.postgresql.mpp import get_mpp
+        return dcs_class(config[name], get_mpp(config))
 
     available_implementations = ', '.join(sorted([n for n, _ in iter_dcs_classes()]))
     raise PatroniFatalException("Can not find suitable configuration of distributed configuration store\n"
@@ -1338,15 +1336,15 @@ class AbstractDCS(abc.ABC):
     _SYNC = 'sync'
     _FAILSAFE = 'failsafe'
 
-    def __init__(self, config: Dict[str, Any]) -> None:
+    def __init__(self, config: Dict[str, Any], mpp: 'AbstractMPP') -> None:
         """Prepare DCS paths, Citus group ID, initial values for state information and processing dependencies.
 
         :ivar config: :class:`dict`, reference to config section of selected DCS.
                       i.e.: ``zookeeper`` for zookeeper, ``etcd`` for etcd, etc...
         """
+        self._mpp = mpp
         self._name = config['name']
         self._base_path = re.sub('/+', '/', '/'.join(['', config.get('namespace', 'service'), config['scope']]))
-        self._citus_group = str(config['group']) if isinstance(config.get('group'), int) else None
         self._set_loop_wait(config.get('loop_wait', 10))
 
         self._ctl = bool(config.get('patronictl', False))
@@ -1359,6 +1357,11 @@ class AbstractDCS(abc.ABC):
         self._last_failsafe: Optional[Dict[str, str]] = {}
         self.event = Event()
 
+    @property
+    def mpp(self) -> 'AbstractMPP':
+        """Get the effective underlying MPP, if any has been configured."""
+        return self._mpp
+
     def client_path(self, path: str) -> str:
         """Construct the absolute key name from appropriate parts for the DCS type.
 
@@ -1367,8 +1370,8 @@ class AbstractDCS(abc.ABC):
         :returns: absolute key name for the current Patroni cluster.
         """
         components = [self._base_path]
-        if self._citus_group:
-            components.append(self._citus_group)
+        if self._mpp.is_enabled():
+            components.append(str(self._mpp.group))
         components.append(path.lstrip('/'))
         return '/'.join(components)
 
@@ -1522,9 +1525,9 @@ class AbstractDCS(abc.ABC):
     def is_citus_coordinator(self) -> bool:
         """:class:`Cluster` instance has a Citus Coordinator group ID.
 
-        :returns: ``True`` if the given node is running as Citus Coordinator (``group=0``).
+        :returns: ``True`` if the given node is running as the MPP Coordinator.
         """
-        return self._citus_group == str(CITUS_COORDINATOR_GROUP_ID)
+        return self._mpp.is_coordinator()
 
     def get_citus_coordinator(self) -> Optional[Cluster]:
         """Load the Patroni cluster for the Citus Coordinator.
@@ -1532,10 +1535,10 @@ class AbstractDCS(abc.ABC):
           .. note::
               This method is only executed on the worker nodes (``group!=0``) to find the coordinator.
 
-        :returns: Select :class:`Cluster` instance associated with the Citus Coordinator group ID.
+        :returns: Select :class:`Cluster` instance associated with the MPP Coordinator group ID.
         """
         try:
-            return self.__get_patroni_cluster(f'{self._base_path}/{CITUS_COORDINATOR_GROUP_ID}/')
+            return self.__get_patroni_cluster(f'{self._base_path}/{self._mpp.coordinator_group_id}/')
         except Exception as e:
             logger.error('Failed to load Citus coordinator cluster from %s: %r', self.__class__.__name__, e)
         return None
@@ -1549,7 +1552,7 @@ class AbstractDCS(abc.ABC):
         groups = self._load_cluster(self._base_path + '/', self._citus_cluster_loader)
         if TYPE_CHECKING:  # pragma: no cover
             assert isinstance(groups, dict)
-        cluster = groups.pop(CITUS_COORDINATOR_GROUP_ID, Cluster.empty())
+        cluster = groups.pop(self._mpp.coordinator_group_id, Cluster.empty())
         cluster.workers.update(groups)
         return cluster
 
diff --git a/patroni/dcs/consul.py b/patroni/dcs/consul.py
index fe66c6d8..19d65306 100644
--- a/patroni/dcs/consul.py
+++ b/patroni/dcs/consul.py
@@ -16,8 +16,9 @@ from urllib.parse import urlencode, urlparse, quote
 from typing import Any, Callable, Dict, List, Mapping, NamedTuple, Optional, Union, Tuple, TYPE_CHECKING
 
 from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \
-    TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
+    TimelineHistory, ReturnFalseException, catch_return_false_exception
 from ..exceptions import DCSError
+from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, parse_bool, Retry, RetryFailedError, split_host_port, uri, USER_AGENT
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
@@ -232,8 +233,8 @@ def service_name_from_scope_name(scope_name: str) -> str:
 
 class Consul(AbstractDCS):
 
-    def __init__(self, config: Dict[str, Any]) -> None:
-        super(Consul, self).__init__(config)
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
+        super(Consul, self).__init__(config, mpp)
         self._base_path = self._base_path[1:]
         self._scope = config['scope']
         self._session = None
@@ -435,7 +436,7 @@ class Consul(AbstractDCS):
         clusters: Dict[int, Dict[str, Cluster]] = defaultdict(dict)
         for node in results or []:
             key = node['Key'][len(path):].split('/', 1)
-            if len(key) == 2 and citus_group_re.match(key[0]):
+            if len(key) == 2 and self._mpp.group_re.match(key[0]):
                 node['Value'] = (node['Value'] or b'').decode('utf-8')
                 clusters[int(key[0])][key[1]] = node
         return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()}
diff --git a/patroni/dcs/etcd.py b/patroni/dcs/etcd.py
index 3be699a6..b9d3e0ab 100644
--- a/patroni/dcs/etcd.py
+++ b/patroni/dcs/etcd.py
@@ -22,8 +22,9 @@ from urllib3 import Timeout
 from urllib3.exceptions import HTTPError, ReadTimeoutError, ProtocolError
 
 from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \
-    TimelineHistory, ReturnFalseException, catch_return_false_exception, citus_group_re
+    TimelineHistory, ReturnFalseException, catch_return_false_exception
 from ..exceptions import DCSError
+from ..postgresql.mpp import AbstractMPP
 from ..request import get as requests_get
 from ..utils import Retry, RetryFailedError, split_host_port, uri, USER_AGENT
 if TYPE_CHECKING:  # pragma: no cover
@@ -470,9 +471,9 @@ class EtcdClient(AbstractEtcdClientWithFailover):
 
 class AbstractEtcd(AbstractDCS):
 
-    def __init__(self, config: Dict[str, Any], client_cls: Type[AbstractEtcdClientWithFailover],
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP, client_cls: Type[AbstractEtcdClientWithFailover],
                  retry_errors_cls: Union[Type[Exception], Tuple[Type[Exception], ...]]) -> None:
-        super(AbstractEtcd, self).__init__(config)
+        super(AbstractEtcd, self).__init__(config, mpp)
         self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
                             retry_exceptions=retry_errors_cls)
         self._ttl = int(config.get('ttl') or 30)
@@ -645,8 +646,8 @@ def catch_etcd_errors(func: Callable[..., Any]) -> Any:
 
 class Etcd(AbstractEtcd):
 
-    def __init__(self, config: Dict[str, Any]) -> None:
-        super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal))
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
+        super(Etcd, self).__init__(config, mpp, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal))
         self.__do_not_watch = False
 
     @property
@@ -726,7 +727,7 @@ class Etcd(AbstractEtcd):
         clusters: Dict[int, Dict[str, etcd.EtcdResult]] = defaultdict(dict)
         for node in result.leaves:
             key = node.key[len(result.key):].lstrip('/').split('/', 1)
-            if len(key) == 2 and citus_group_re.match(key[0]):
+            if len(key) == 2 and self._mpp.group_re.match(key[0]):
                 clusters[int(key[0])][key[1]] = node
         return {group: self._cluster_from_nodes(result.etcd_index, nodes) for group, nodes in clusters.items()}
 
diff --git a/patroni/dcs/etcd3.py b/patroni/dcs/etcd3.py
index ea7e52f2..7cc2a115 100644
--- a/patroni/dcs/etcd3.py
+++ b/patroni/dcs/etcd3.py
@@ -16,9 +16,10 @@ from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, Iterator, List, Optional, Tuple, Type, TYPE_CHECKING, Union
 
 from . import ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \
-    TimelineHistory, catch_return_false_exception, citus_group_re
+    TimelineHistory, catch_return_false_exception
 from .etcd import AbstractEtcdClientWithFailover, AbstractEtcd, catch_etcd_errors, DnsCachingResolver, Retry
 from ..exceptions import DCSError, PatroniException
+from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, enable_keepalive, iter_response_objects, RetryFailedError, USER_AGENT
 
 logger = logging.getLogger(__name__)
@@ -671,8 +672,9 @@ class PatroniEtcd3Client(Etcd3Client):
 
 class Etcd3(AbstractEtcd):
 
-    def __init__(self, config: Dict[str, Any]) -> None:
-        super(Etcd3, self).__init__(config, PatroniEtcd3Client, (DeadlineExceeded, Unavailable, FailedPrecondition))
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
+        super(Etcd3, self).__init__(config, mpp, PatroniEtcd3Client,
+                                    (DeadlineExceeded, Unavailable, FailedPrecondition))
         self.__do_not_watch = False
         self._lease = None
         self._last_lease_refresh = 0
@@ -796,7 +798,7 @@ class Etcd3(AbstractEtcd):
         path = self._base_path + '/'
         for node in self._client.get_cluster(path):
             key = node['key'][len(path):].split('/', 1)
-            if len(key) == 2 and citus_group_re.match(key[0]):
+            if len(key) == 2 and self._mpp.group_re.match(key[0]):
                 clusters[int(key[0])][key[1]] = node
         return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()}
 
diff --git a/patroni/dcs/exhibitor.py b/patroni/dcs/exhibitor.py
index 2b06073b..03d23575 100644
--- a/patroni/dcs/exhibitor.py
+++ b/patroni/dcs/exhibitor.py
@@ -7,6 +7,7 @@ from typing import Any, Callable, Dict, List, Union
 
 from . import Cluster
 from .zookeeper import ZooKeeper
+from ..postgresql.mpp import AbstractMPP
 from ..request import get as requests_get
 from ..utils import uri
 
@@ -66,10 +67,10 @@ class ExhibitorEnsembleProvider(object):
 
 class Exhibitor(ZooKeeper):
 
-    def __init__(self, config: Dict[str, Any]) -> None:
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
         interval = config.get('poll_interval', 300)
         self._ensemble_provider = ExhibitorEnsembleProvider(config['hosts'], config['port'], poll_interval=interval)
-        super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts})
+        super(Exhibitor, self).__init__({**config, 'hosts': self._ensemble_provider.zookeeper_hosts}, mpp)
 
     def _load_cluster(
             self, path: str, loader: Callable[[str], Union[Cluster, Dict[int, Cluster]]]
diff --git a/patroni/dcs/kubernetes.py b/patroni/dcs/kubernetes.py
index aee87bd3..343b496c 100644
--- a/patroni/dcs/kubernetes.py
+++ b/patroni/dcs/kubernetes.py
@@ -19,9 +19,9 @@ from urllib3.exceptions import HTTPError
 from threading import Condition, Lock, Thread
 from typing import Any, Callable, Collection, Dict, List, Optional, Tuple, Type, Union, TYPE_CHECKING
 
-from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, \
-    TimelineHistory, CITUS_COORDINATOR_GROUP_ID, citus_group_re
+from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, Status, SyncState, TimelineHistory
 from ..exceptions import DCSError
+from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare, iter_response_objects, keepalive_socket_options, \
     Retry, RetryFailedError, tzutc, uri, USER_AGENT
 if TYPE_CHECKING:  # pragma: no cover
@@ -748,7 +748,7 @@ class Kubernetes(AbstractDCS):
 
     _CITUS_LABEL = 'citus-group'
 
-    def __init__(self, config: Dict[str, Any]) -> None:
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
         self._labels = deepcopy(config['labels'])
         self._labels[config.get('scope_label', 'cluster-name')] = config['scope']
         self._label_selector = ','.join('{0}={1}'.format(k, v) for k, v in self._labels.items())
@@ -759,9 +759,9 @@ class Kubernetes(AbstractDCS):
         self._standby_leader_label_value = config.get('standby_leader_label_value', 'master')
         self._tmp_role_label = config.get('tmp_role_label')
         self._ca_certs = os.environ.get('PATRONI_KUBERNETES_CACERT', config.get('cacert')) or SERVICE_CERT_FILENAME
-        super(Kubernetes, self).__init__({**config, 'namespace': ''})
-        if self._citus_group:
-            self._labels[self._CITUS_LABEL] = self._citus_group
+        super(Kubernetes, self).__init__({**config, 'namespace': ''}, mpp)
+        if self._mpp.is_enabled():
+            self._labels[self._CITUS_LABEL] = str(self._mpp.group)
 
         self._retry = Retry(deadline=config['retry_timeout'], max_delay=1, max_tries=-1,
                             retry_exceptions=KubernetesRetriableException)
@@ -944,12 +944,12 @@ class Kubernetes(AbstractDCS):
 
         for name, pod in path['pods'].items():
             group = pod.metadata.labels.get(self._CITUS_LABEL)
-            if group and citus_group_re.match(group):
+            if group and self._mpp.group_re.match(group):
                 clusters[group]['pods'][name] = pod
 
         for name, kind in path['nodes'].items():
             group = kind.metadata.labels.get(self._CITUS_LABEL)
-            if group and citus_group_re.match(group):
+            if group and self._mpp.group_re.match(group):
                 clusters[group]['nodes'][name] = kind
         return {int(group): self._cluster_from_nodes(group, value['nodes'], value['pods'].values())
                 for group, value in clusters.items()}
@@ -976,12 +976,12 @@ class Kubernetes(AbstractDCS):
     def _load_cluster(
             self, path: str, loader: Callable[[Any], Union[Cluster, Dict[int, Cluster]]]
     ) -> Union[Cluster, Dict[int, Cluster]]:
-        group = self._citus_group if path == self.client_path('') else None
+        group = str(self._mpp.group) if self._mpp.is_enabled() and path == self.client_path('') else None
         return self.__load_cluster(group, loader)
 
     def get_citus_coordinator(self) -> Optional[Cluster]:
         try:
-            ret = self.__load_cluster(str(CITUS_COORDINATOR_GROUP_ID), self._cluster_loader)
+            ret = self.__load_cluster(str(self._mpp.coordinator_group_id), self._cluster_loader)
             if TYPE_CHECKING:  # pragma: no cover
                 assert isinstance(ret, Cluster)
             return ret
diff --git a/patroni/dcs/raft.py b/patroni/dcs/raft.py
index 98c48f44..0528cfb0 100644
--- a/patroni/dcs/raft.py
+++ b/patroni/dcs/raft.py
@@ -12,9 +12,9 @@ from pysyncobj.transport import TCPTransport, CONNECTION_STATE
 from pysyncobj.utility import TcpUtility
 from typing import Any, Callable, Collection, Dict, List, Optional, Set, Union, TYPE_CHECKING
 
-from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \
-    TimelineHistory, citus_group_re
+from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory
 from ..exceptions import DCSError
+from ..postgresql.mpp import AbstractMPP
 from ..utils import validate_directory
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
@@ -285,8 +285,8 @@ class KVStoreTTL(DynMemberSyncObj):
 
 class Raft(AbstractDCS):
 
-    def __init__(self, config: Dict[str, Any]) -> None:
-        super(Raft, self).__init__(config)
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
+        super(Raft, self).__init__(config, mpp)
         self._ttl = int(config.get('ttl') or 30)
 
         ready_event = threading.Event()
@@ -387,7 +387,7 @@ class Raft(AbstractDCS):
         response = self._sync_obj.get(path, recursive=True)
         for key, value in (response or {}).items():
             key = key[len(path):].split('/', 1)
-            if len(key) == 2 and citus_group_re.match(key[0]):
+            if len(key) == 2 and self._mpp.group_re.match(key[0]):
                 clusters[int(key[0])][key[1]] = value
         return {group: self._cluster_from_nodes(nodes) for group, nodes in clusters.items()}
 
diff --git a/patroni/dcs/zookeeper.py b/patroni/dcs/zookeeper.py
index 3704b579..6bf77ae4 100644
--- a/patroni/dcs/zookeeper.py
+++ b/patroni/dcs/zookeeper.py
@@ -12,9 +12,9 @@ from kazoo.retry import RetryFailedError
 from kazoo.security import ACL, make_acl
 from typing import Any, Callable, Dict, List, Optional, Union, Tuple, TYPE_CHECKING
 
-from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, \
-    TimelineHistory, citus_group_re
+from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, Status, SyncState, TimelineHistory
 from ..exceptions import DCSError
+from ..postgresql.mpp import AbstractMPP
 from ..utils import deep_compare
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
@@ -87,8 +87,8 @@ class PatroniKazooClient(KazooClient):
 
 class ZooKeeper(AbstractDCS):
 
-    def __init__(self, config: Dict[str, Any]) -> None:
-        super(ZooKeeper, self).__init__(config)
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
+        super(ZooKeeper, self).__init__(config, mpp)
 
         hosts: Union[str, List[str]] = config.get('hosts', [])
         if isinstance(hosts, list):
@@ -261,7 +261,7 @@ class ZooKeeper(AbstractDCS):
     def _citus_cluster_loader(self, path: str) -> Dict[int, Cluster]:
         ret: Dict[int, Cluster] = {}
         for node in self.get_children(path):
-            if citus_group_re.match(node):
+            if self._mpp.group_re.match(node):
                 ret[int(node)] = self._cluster_loader(path + node + '/')
         return ret
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 0d3e05a4..dea78163 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -332,7 +332,7 @@ class Ha(object):
             if coordinator and coordinator.leader and coordinator.leader.conn_url:
                 try:
                     data = {'type': event,
-                            'group': self.state_handler.citus_handler.group(),
+                            'group': self.state_handler.citus_handler.group,
                             'leader': self.state_handler.name,
                             'timeout': self.dcs.ttl,
                             'cooldown': self.patroni.config['retry_timeout']}
@@ -847,7 +847,7 @@ class Ha(object):
             self.state_handler.set_role('master')
             self.process_sync_replication()
             self.update_cluster_history()
-            self.state_handler.citus_handler.sync_pg_dist_node(self.cluster)
+            self.state_handler.citus_handler.sync_meta_data(self.cluster)
             return message
         elif self.state_handler.role in ('master', 'promoted', 'primary'):
             self.process_sync_replication()
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index c87c29fa..f6e3ab54 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -19,8 +19,8 @@ from .callback_executor import CallbackAction, CallbackExecutor
 from .cancellable import CancellableSubprocess
 from .config import ConfigHandler, mtime
 from .connection import ConnectionPool, get_connection_cursor
-from .citus import CitusHandler
 from .misc import parse_history, parse_lsn, postgres_major_version_to_int
+from .mpp import AbstractMPP
 from .postmaster import PostmasterProcess
 from .slots import SlotsHandler
 from .sync import SyncHandler
@@ -63,7 +63,7 @@ class Postgresql(object):
               "pg_catalog.pg_{0}_{1}_diff(COALESCE(pg_catalog.pg_last_{0}_receive_{1}(), '0/0'), '0/0')::bigint, "
               "pg_catalog.pg_is_in_recovery() AND pg_catalog.pg_is_{0}_replay_paused()")
 
-    def __init__(self, config: Dict[str, Any]) -> None:
+    def __init__(self, config: Dict[str, Any], mpp: AbstractMPP) -> None:
         self.name: str = config['name']
         self.scope: str = config['scope']
         self._data_dir: str = config['data_dir']
@@ -80,7 +80,7 @@ class Postgresql(object):
         self._pending_restart = False
         self.connection_pool = ConnectionPool()
         self._connection = self.connection_pool.get('heartbeat')
-        self.citus_handler = CitusHandler(self, config.get('citus'))
+        self.citus_handler = mpp.get_handler_impl(self)
         self.config = ConfigHandler(self, config)
         self.config.check_directories()
 
diff --git a/patroni/postgresql/mpp/__init__.py b/patroni/postgresql/mpp/__init__.py
new file mode 100644
index 00000000..3494793b
--- /dev/null
+++ b/patroni/postgresql/mpp/__init__.py
@@ -0,0 +1,296 @@
+"""Abstract classes for MPP handler.
+
+MPP stands for Massively Parallel Processing, and Citus belongs to this architecture. Currently, Citus is the only
+supported MPP cluster. However, we may consider adapting other databases such as TimescaleDB, GPDB, etc. into Patroni.
+"""
+import abc
+
+from typing import Any, Dict, Iterator, Optional, Union, Tuple, Type, TYPE_CHECKING
+
+from ...dcs import Cluster
+from ...dynamic_loader import iter_classes
+from ...exceptions import PatroniException
+
+if TYPE_CHECKING:  # pragma: no cover
+    from .. import Postgresql
+    from ...config import Config
+
+
+class AbstractMPP(abc.ABC):
+    """An abstract class which should be passed to :class:`AbstractDCS`.
+
+    .. note::
+        We create :class:`AbstractMPP` and :class:`AbstractMPPHandler` to solve the chicken-egg initialization problem.
+        When initializing DCS, we dynamically create an object implementing :class:`AbstractMPP`, later this object is
+        used to instantiate an object implementing :class:`AbstractMPPHandler`.
+    """
+
+    group_re: Any  # re.Pattern[str]
+
+    def __init__(self, config: Dict[str, Union[str, int]]) -> None:
+        """Init method for :class:`AbstractMPP`.
+
+        :param config: configuration of MPP section.
+        """
+        self._config = config
+
+    def is_enabled(self) -> bool:
+        """Check if MPP is enabled for a given MPP.
+
+        .. note::
+            We just check that the :attr:`_config` object isn't empty and expect
+            it to be empty only in case of :class:`Null`.
+
+        :returns: ``True`` if MPP is enabled, otherwise ``False``.
+        """
+        return bool(self._config)
+
+    @staticmethod
+    @abc.abstractmethod
+    def validate_config(config: Any) -> bool:
+        """Check whether provided config is good for a given MPP.
+
+        :param config: configuration of MPP section.
+
+        :returns: ``True`` is config passes validation, otherwise ``False``.
+        """
+
+    @property
+    @abc.abstractmethod
+    def group(self) -> Any:
+        """The group for a given MPP implementation."""
+
+    @property
+    @abc.abstractmethod
+    def coordinator_group_id(self) -> Any:
+        """The group id of the coordinator PostgreSQL cluster."""
+
+    def is_coordinator(self) -> bool:
+        """Check whether this node is running in the coordinator PostgreSQL cluster.
+
+        :returns: ``True`` if MPP is enabled and the group id of this node
+                  matches with the :attr:`coordinator_group_id`, otherwise ``False``.
+        """
+        return self.is_enabled() and self.group == self.coordinator_group_id
+
+    def is_worker(self) -> bool:
+        """Check whether this node is running as a MPP worker PostgreSQL cluster.
+
+        :returns: ``True`` if MPP is enabled and this node is known to be not running
+                  as the coordinator PostgreSQL cluster, otherwise ``False``.
+        """
+        return self.is_enabled() and not self.is_coordinator()
+
+    def _get_handler_cls(self) -> Iterator[Type['AbstractMPPHandler']]:
+        """Find Handler classes inherited from a class type of this object.
+
+        :yields: handler classes for this object.
+        """
+        for cls in self.__class__.__subclasses__():
+            if issubclass(cls, AbstractMPPHandler) and cls.__name__.startswith(self.__class__.__name__):
+                yield cls
+
+    def get_handler_impl(self, postgresql: 'Postgresql') -> 'AbstractMPPHandler':
+        """Find and instantiate Handler implementation of this object.
+
+        :param postgresql: a reference to :class:`Postgresql` object.
+
+        :raises:
+            :exc:`PatroniException`: if the Handler class haven't been found.
+
+        :returns: an instantiated class that implements Handler for this object.
+        """
+        for cls in self._get_handler_cls():
+            return cls(postgresql, self._config)
+        raise PatroniException(f'Failed to initialize {self.__class__.__name__}Handler object')
+
+
+class AbstractMPPHandler(AbstractMPP):
+    """An abstract class which defines interfaces that should be implemented by real handlers."""
+
+    def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None:
+        """Init method for :class:`AbstractMPPHandler`.
+
+        :param postgresql: a reference to :class:`Postgresql` object.
+        :param config: configuration of MPP section.
+        """
+        super().__init__(config)
+        self._postgresql = postgresql
+
+    @abc.abstractmethod
+    def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None:
+        """Handle an event sent from a worker node.
+
+        :param cluster: the currently known cluster state from DCS.
+        :param event: the event to be handled.
+        """
+
+    @abc.abstractmethod
+    def sync_meta_data(self, cluster: Cluster) -> None:
+        """Sync meta data on the coordinator.
+
+        :param cluster: the currently known cluster state from DCS.
+        """
+
+    @abc.abstractmethod
+    def on_demote(self) -> None:
+        """On demote handler.
+
+        Is called when the primary was demoted.
+        """
+
+    @abc.abstractmethod
+    def schedule_cache_rebuild(self) -> None:
+        """Cache rebuild handler.
+
+        Is called to notify handler that it has to refresh its metadata cache from the database.
+        """
+
+    @abc.abstractmethod
+    def bootstrap(self) -> None:
+        """Bootstrap handler.
+
+        Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method).
+        """
+
+    @abc.abstractmethod
+    def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None:
+        """Adjust GUCs in the current PostgreSQL configuration.
+
+        :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value.
+        """
+
+    @abc.abstractmethod
+    def ignore_replication_slot(self, slot: Dict[str, str]) -> bool:
+        """Check whether provided replication *slot* existing in the database should not be removed.
+
+        .. note::
+            MPP database may create replication slots for its own use, for example to migrate data between workers
+            using logical replication, and we don't want to suddenly drop them.
+
+        :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and
+                     ``plugin``.
+
+        :returns: ``True`` if the replication slots should not be removed, otherwise ``False``.
+        """
+
+
+class Null(AbstractMPP):
+    """Dummy implementation of :class:`AbstractMPP`."""
+
+    def __init__(self) -> None:
+        """Init method for :class:`Null`."""
+        super().__init__({})
+
+    @staticmethod
+    def validate_config(config: Any) -> bool:
+        """Check whether provided config is good for :class:`Null`.
+
+        :returns: always ``True``.
+        """
+        return True
+
+    @property
+    def group(self) -> None:
+        """The group for :class:`Null`.
+
+        :returns: always ``None``.
+        """
+        return None
+
+    @property
+    def coordinator_group_id(self) -> None:
+        """The group id of the coordinator PostgreSQL cluster.
+
+        :returns: always ``None``.
+        """
+        return None
+
+
+class NullHandler(Null, AbstractMPPHandler):
+    """Dummy implementation of :class:`AbstractMPPHandler`."""
+
+    def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None:
+        """Init method for :class:`NullHandler`.
+
+        :param postgresql: a reference to :class:`Postgresql` object.
+        :param config: configuration of MPP section.
+        """
+        AbstractMPPHandler.__init__(self, postgresql, config)
+
+    def handle_event(self, cluster: Cluster, event: Dict[str, Any]) -> None:
+        """Handle an event sent from a worker node.
+
+        :param cluster: the currently known cluster state from DCS.
+        :param event: the event to be handled.
+        """
+
+    def sync_meta_data(self, cluster: Cluster) -> None:
+        """Sync meta data on the coordinator.
+
+        :param cluster: the currently known cluster state from DCS.
+        """
+
+    def on_demote(self) -> None:
+        """On demote handler.
+
+        Is called when the primary was demoted.
+        """
+
+    def schedule_cache_rebuild(self) -> None:
+        """Cache rebuild handler.
+
+        Is called to notify handler that it has to refresh its metadata cache from the database.
+        """
+
+    def bootstrap(self) -> None:
+        """Bootstrap handler.
+
+        Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method).
+        """
+
+    def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None:
+        """Adjust GUCs in the current PostgreSQL configuration.
+
+        :param parameters: dictionary of GUCs, with key as GUC name and corresponding value as current GUC value.
+        """
+
+    def ignore_replication_slot(self, slot: Dict[str, str]) -> bool:
+        """Check whether provided replication *slot* existing in the database should not be removed.
+
+        .. note::
+            MPP database may create replication slots for its own use, for example to migrate data between workers
+            using logical replication, and we don't want to suddenly drop them.
+
+        :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and
+                     ``plugin``.
+
+        :returns: always ``False``.
+        """
+        return False
+
+
+def iter_mpp_classes(
+        config: Optional[Union['Config', Dict[str, Any]]] = None
+) -> Iterator[Tuple[str, Type[AbstractMPP]]]:
+    """Attempt to import MPP modules that are present in the given configuration.
+
+    :param config: configuration information with possible MPP names as keys. If given, only attempt to import MPP
+                   modules defined in the configuration. Else, if ``None``, attempt to import any supported MPP module.
+
+    :yields: tuples, each containing the module ``name`` and the imported MPP class object.
+    """
+    yield from iter_classes(__package__, AbstractMPP, config)
+
+
+def get_mpp(config: Union['Config', Dict[str, Any]]) -> AbstractMPP:
+    """Attempt to load and instantiate a MPP module from known available implementations.
+
+    :param config: object or dictionary with Patroni configuration.
+
+    :returns: The successfully loaded MPP or fallback to :class:`Null`.
+    """
+    for name, mpp_class in iter_mpp_classes(config):
+        if mpp_class.validate_config(config[name]):
+            return mpp_class(config[name])
+    return Null()
diff --git a/patroni/postgresql/citus.py b/patroni/postgresql/mpp/citus.py
similarity index 84%
rename from patroni/postgresql/citus.py
rename to patroni/postgresql/mpp/citus.py
index b50dc1d0..a17d1435 100644
--- a/patroni/postgresql/citus.py
+++ b/patroni/postgresql/mpp/citus.py
@@ -6,12 +6,15 @@ from threading import Condition, Event, Thread
 from urllib.parse import urlparse
 from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING
 
-from ..dcs import CITUS_COORDINATOR_GROUP_ID, Cluster
-from ..psycopg import connect, quote_ident
+from . import AbstractMPP, AbstractMPPHandler
+from ...dcs import Cluster
+from ...psycopg import connect, quote_ident
+from ...utils import parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
-    from . import Postgresql
+    from .. import Postgresql
 
+CITUS_COORDINATOR_GROUP_ID = 0
 CITUS_SLOT_NAME_RE = re.compile(r'^citus_shard_(move|split)_slot(_[1-9][0-9]*){2,3}$')
 logger = logging.getLogger(__name__)
 
@@ -63,13 +66,45 @@ class PgDistNode(object):
         return str(self)
 
 
-class CitusHandler(Thread):
+class Citus(AbstractMPP):
 
-    def __init__(self, postgresql: 'Postgresql', config: Optional[Dict[str, Union[str, int]]]) -> None:
-        super(CitusHandler, self).__init__()
+    group_re = re.compile('^(0|[1-9][0-9]*)$')
+
+    @staticmethod
+    def validate_config(config: Union[Any, Dict[str, Union[str, int]]]) -> bool:
+        """Check whether provided config is good for a given MPP.
+
+        :param config: configuration of ``citus`` MPP section.
+
+        :returns: ``True`` is config passes validation, otherwise ``False``.
+        """
+        return isinstance(config, dict) \
+            and isinstance(config.get('database'), str) \
+            and parse_int(config.get('group')) is not None
+
+    @property
+    def group(self) -> int:
+        """The group of this Citus node."""
+        return int(self._config['group'])
+
+    @property
+    def coordinator_group_id(self) -> int:
+        """The group id of the Citus coordinator PostgreSQL cluster."""
+        return CITUS_COORDINATOR_GROUP_ID
+
+
+class CitusHandler(Citus, AbstractMPPHandler, Thread):
+    """Define the interfaces for handling an underlying Citus cluster."""
+
+    def __init__(self, postgresql: 'Postgresql', config: Dict[str, Union[str, int]]) -> None:
+        """"Initialize a new instance of :class:`CitusHandler`.
+
+        :param postgresql: the Postgres node.
+        :param config: the ``citus`` MPP config section.
+        """
+        Thread.__init__(self)
+        AbstractMPPHandler.__init__(self, postgresql, config)
         self.daemon = True
-        self._postgresql = postgresql
-        self._config = config
         if config:
             self._connection = postgresql.connection_pool.get(
                 'citus', {'dbname': config['database'],
@@ -81,19 +116,11 @@ class CitusHandler(Thread):
         self._condition = Condition()  # protects _pg_dist_node, _tasks, _in_flight, and _schedule_load_pg_dist_node
         self.schedule_cache_rebuild()
 
-    def is_enabled(self) -> bool:
-        return isinstance(self._config, dict)
-
-    def group(self) -> Optional[int]:
-        return int(self._config['group']) if isinstance(self._config, dict) else None
-
-    def is_coordinator(self) -> bool:
-        return self.is_enabled() and self.group() == CITUS_COORDINATOR_GROUP_ID
-
-    def is_worker(self) -> bool:
-        return self.is_enabled() and not self.is_coordinator()
-
     def schedule_cache_rebuild(self) -> None:
+        """Cache rebuild handler.
+
+        Is called to notify handler that it has to refresh its metadata cache from the database.
+        """
         with self._condition:
             self._schedule_load_pg_dist_node = True
 
@@ -134,8 +161,8 @@ class CitusHandler(Thread):
             self._pg_dist_node = {r[1]: PgDistNode(r[1], r[2], r[3], 'after_promote', r[0]) for r in rows}
         return True
 
-    def sync_pg_dist_node(self, cluster: Cluster) -> None:
-        """Maintain the `pg_dist_node` from the coordinator leader every heartbeat loop.
+    def sync_meta_data(self, cluster: Cluster) -> None:
+        """Maintain the ``pg_dist_node`` from the coordinator leader every heartbeat loop.
 
         We can't always rely on REST API calls from worker nodes in order
         to maintain `pg_dist_node`, therefore at least once per heartbeat
@@ -296,16 +323,16 @@ class CitusHandler(Thread):
         with self._condition:
             i = self.find_task_by_group(task.group)
 
-            # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_pg_dist_node().
+            # The `PgDistNode.timeout` == None is an indicator that it was scheduled from the sync_meta_data().
             if task.timeout is None:
                 # We don't want to override the already existing task created from REST API.
                 if i is not None and self._tasks[i].timeout is not None:
                     return False
 
                 # There is a little race condition with tasks created from REST API - the call made "before" the member
-                # key is updated in DCS. Therefore it is possible that :func:`sync_pg_dist_node` will try to create a
-                # task based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout.
-                # Only when the timeout is reached new tasks could be scheduled from sync_pg_dist_node()
+                # key is updated in DCS. Therefore it is possible that :func:`sync_meta_data` will try to create a task
+                # based on the outdated values of "state"/"role". To solve it we introduce an artificial timeout.
+                # Only when the timeout is reached new tasks could be scheduled from sync_meta_data()
                 if self._in_flight and self._in_flight.group == task.group and self._in_flight.timeout is not None\
                         and self._in_flight.deadline > time.time():
                     return False
@@ -353,9 +380,10 @@ class CitusHandler(Thread):
             task.wait()
 
     def bootstrap(self) -> None:
-        if not isinstance(self._config, dict):  # self.is_enabled()
-            return
+        """Bootstrap handler.
 
+        Is called when the new cluster is initialized (through ``initdb`` or a custom bootstrap method).
+        """
         conn_kwargs = {**self._postgresql.connection_pool.conn_kwargs,
                        'options': '-c synchronous_commit=local -c statement_timeout=0'}
         if self._config['database'] != self._postgresql.database:
@@ -388,9 +416,10 @@ class CitusHandler(Thread):
             conn.close()
 
     def adjust_postgres_gucs(self, parameters: Dict[str, Any]) -> None:
-        if not self.is_enabled():
-            return
+        """Adjust GUCs in the current PostgreSQL configuration.
 
+        :param parameters: dictionary of GUCs, with key as GUC name and the corresponding value as current GUC value.
+        """
         # citus extension must be on the first place in shared_preload_libraries
         shared_preload_libraries = list(filter(
             lambda el: el and el != 'citus',
@@ -408,8 +437,18 @@ class CitusHandler(Thread):
         parameters['citus.local_hostname'] = self._postgresql.connection_pool.conn_kwargs.get('host', 'localhost')
 
     def ignore_replication_slot(self, slot: Dict[str, str]) -> bool:
-        if isinstance(self._config, dict) and self._postgresql.is_primary() and\
-                slot['type'] == 'logical' and slot['database'] == self._config['database']:
+        """Check whether provided replication *slot* existing in the database should not be removed.
+
+        .. note::
+            MPP database may create replication slots for its own use, for example to migrate data between workers
+            using logical replication, and we don't want to suddenly drop them.
+
+        :param slot: dictionary containing the replication slot settings, like ``name``, ``database``, ``type``, and
+                     ``plugin``.
+
+        :returns: ``True`` if the replication slots should not be removed, otherwise ``False``.
+        """
+        if self._postgresql.is_primary() and slot['type'] == 'logical' and slot['database'] == self._config['database']:
             m = CITUS_SLOT_NAME_RE.match(slot['name'])
             return bool(m and {'move': 'pgoutput', 'split': 'citus'}.get(m.group(1)) == slot['plugin'])
         return False
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index fb9448cd..7f7fd294 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -291,7 +291,7 @@ class SlotsHandler:
         :param name: name of the slot to ignore
 
         :returns: ``True`` if slot *name* matches any slot specified in ``ignore_slots`` configuration,
-                 otherwise will pass through and return result of :meth:`CitusHandler.ignore_replication_slot`.
+                 otherwise will pass through and return result of :meth:`AbstractMPPHandler.ignore_replication_slot`.
         """
         slot = self._replication_slots[name]
         if cluster.config:
diff --git a/tests/__init__.py b/tests/__init__.py
index b013e4e1..986bd88b 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -12,6 +12,7 @@ import patroni.psycopg as psycopg
 from patroni.dcs import Leader, Member
 from patroni.postgresql import Postgresql
 from patroni.postgresql.config import ConfigHandler
+from patroni.postgresql.mpp import get_mpp
 from patroni.utils import RetryFailedError, tzutc
 
 
@@ -252,23 +253,24 @@ class PostgresInit(unittest.TestCase):
     @patch.object(Postgresql, 'get_postgres_role_from_data_directory', Mock(return_value='primary'))
     def setUp(self):
         data_dir = os.path.join('data', 'test0')
-        self.p = Postgresql({'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir,
-                             'config_dir': data_dir, 'retry_timeout': 10,
-                             'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'),
-                             'listen': '127.0.0.2, 127.0.0.3:5432',
-                             'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433',
-                             'authentication': {'superuser': {'username': 'foo', 'password': 'test'},
-                                                'replication': {'username': '', 'password': 'rep-pass'},
-                                                'rewind': {'username': 'rewind', 'password': 'test'}},
-                             'remove_data_directory_on_rewind_failure': True,
-                             'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True,
-                             'parameters': self._PARAMETERS,
-                             'recovery_conf': {'foo': 'bar'},
-                             'pg_hba': ['host all all 0.0.0.0/0 md5'],
-                             'pg_ident': ['krb realm postgres'],
-                             'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true',
-                                           'on_restart': 'true', 'on_role_change': 'true'},
-                             'citus': {'group': 0, 'database': 'citus'}})
+        config = {'name': 'postgresql0', 'scope': 'batman', 'data_dir': data_dir,
+                  'config_dir': data_dir, 'retry_timeout': 10,
+                  'krbsrvname': 'postgres', 'pgpass': os.path.join(data_dir, 'pgpass0'),
+                  'listen': '127.0.0.2, 127.0.0.3:5432',
+                  'connect_address': '127.0.0.2:5432', 'proxy_address': '127.0.0.2:5433',
+                  'authentication': {'superuser': {'username': 'foo', 'password': 'test'},
+                                     'replication': {'username': '', 'password': 'rep-pass'},
+                                     'rewind': {'username': 'rewind', 'password': 'test'}},
+                  'remove_data_directory_on_rewind_failure': True,
+                  'use_pg_rewind': True, 'pg_ctl_timeout': 'bla', 'use_unix_socket': True,
+                  'parameters': self._PARAMETERS,
+                  'recovery_conf': {'foo': 'bar'},
+                  'pg_hba': ['host all all 0.0.0.0/0 md5'],
+                  'pg_ident': ['krb realm postgres'],
+                  'callbacks': {'on_start': 'true', 'on_stop': 'true', 'on_reload': 'true',
+                                'on_restart': 'true', 'on_role_change': 'true'},
+                  'citus': {'group': 0, 'database': 'citus'}}
+        self.p = Postgresql(config, get_mpp(config))
 
 
 class BaseTestPostgresql(PostgresInit):
diff --git a/tests/test_citus.py b/tests/test_citus.py
index 7279893e..dbf6d9cf 100644
--- a/tests/test_citus.py
+++ b/tests/test_citus.py
@@ -1,12 +1,12 @@
 import time
 from mock import Mock, patch
-from patroni.postgresql.citus import CitusHandler
+from patroni.postgresql.mpp.citus import CitusHandler
 
 from . import BaseTestPostgresql, MockCursor, psycopg_connect, SleepException
 from .test_ha import get_cluster_initialized_with_leader
 
 
-@patch('patroni.postgresql.citus.Thread', Mock())
+@patch('patroni.postgresql.mpp.citus.Thread', Mock())
 @patch('patroni.psycopg.connect', psycopg_connect)
 class TestCitus(BaseTestPostgresql):
 
@@ -17,9 +17,9 @@ class TestCitus(BaseTestPostgresql):
         self.cluster.workers[1] = self.cluster
 
     @patch('time.time', Mock(side_effect=[100, 130, 160, 190, 220, 250, 280, 310, 340, 370]))
-    @patch('patroni.postgresql.citus.logger.exception', Mock(side_effect=SleepException))
-    @patch('patroni.postgresql.citus.logger.warning')
-    @patch('patroni.postgresql.citus.PgDistNode.wait', Mock())
+    @patch('patroni.postgresql.mpp.citus.logger.exception', Mock(side_effect=SleepException))
+    @patch('patroni.postgresql.mpp.citus.logger.warning')
+    @patch('patroni.postgresql.mpp.citus.PgDistNode.wait', Mock())
     @patch.object(CitusHandler, 'is_alive', Mock(return_value=True))
     def test_run(self, mock_logger_warning):
         # `before_demote` or `before_promote` REST API calls starting a
@@ -39,10 +39,10 @@ class TestCitus(BaseTestPostgresql):
 
     @patch.object(CitusHandler, 'is_alive', Mock(return_value=False))
     @patch.object(CitusHandler, 'start', Mock())
-    def test_sync_pg_dist_node(self):
+    def test_sync_meta_data(self):
         with patch.object(CitusHandler, 'is_enabled', Mock(return_value=False)):
-            self.c.sync_pg_dist_node(self.cluster)
-        self.c.sync_pg_dist_node(self.cluster)
+            self.c.sync_meta_data(self.cluster)
+        self.c.sync_meta_data(self.cluster)
 
     def test_handle_event(self):
         self.c.handle_event(self.cluster, {})
@@ -51,22 +51,22 @@ class TestCitus(BaseTestPostgresql):
                                                'leader': 'leader', 'timeout': 30, 'cooldown': 10})
 
     def test_add_task(self):
-        with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
-                patch('patroni.postgresql.citus.urlparse', Mock(side_effect=Exception)):
+        with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \
+                patch('patroni.postgresql.mpp.citus.urlparse', Mock(side_effect=Exception)):
             self.c.add_task('', 1, None)
             mock_logger.assert_called_once()
 
-        with patch('patroni.postgresql.citus.logger.debug') as mock_logger:
+        with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger:
             self.c.add_task('before_demote', 1, 'postgres://host:5432/postgres', 30)
             mock_logger.assert_called_once()
             self.assertTrue(mock_logger.call_args[0][0].startswith('Adding the new task:'))
 
-        with patch('patroni.postgresql.citus.logger.debug') as mock_logger:
+        with patch('patroni.postgresql.mpp.citus.logger.debug') as mock_logger:
             self.c.add_task('before_promote', 1, 'postgres://host:5432/postgres', 30)
             mock_logger.assert_called_once()
             self.assertTrue(mock_logger.call_args[0][0].startswith('Overriding existing task:'))
 
-        # add_task called from sync_pg_dist_node should not override already scheduled or in flight task until deadline
+        # add_task called from sync_meta_data should not override already scheduled or in flight task until deadline
         self.assertIsNotNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres', 30))
         self.assertIsNone(self.c.add_task('after_promote', 1, 'postgres://host:5432/postgres'))
         self.c._in_flight = self.c._tasks.pop()
@@ -106,7 +106,7 @@ class TestCitus(BaseTestPostgresql):
         self.c.process_tasks()
 
         self.c.add_task('after_promote', 0, 'postgres://host3:5432/postgres')
-        with patch('patroni.postgresql.citus.logger.error') as mock_logger, \
+        with patch('patroni.postgresql.mpp.citus.logger.error') as mock_logger, \
                 patch.object(CitusHandler, 'query', Mock(side_effect=Exception)):
             self.c.process_tasks()
             mock_logger.assert_called_once()
@@ -115,7 +115,7 @@ class TestCitus(BaseTestPostgresql):
     def test_on_demote(self):
         self.c.on_demote()
 
-    @patch('patroni.postgresql.citus.logger.error')
+    @patch('patroni.postgresql.mpp.citus.logger.error')
     @patch.object(MockCursor, 'execute', Mock(side_effect=Exception))
     def test_load_pg_dist_node(self, mock_logger):
         # load_pg_dist_node() triggers, query fails and exception is property handled
@@ -140,10 +140,6 @@ class TestCitus(BaseTestPostgresql):
         self.assertEqual(parameters['wal_level'], 'logical')
         self.assertEqual(parameters['citus.local_hostname'], '/tmp')
 
-    def test_bootstrap(self):
-        self.c._config = None
-        self.c.bootstrap()
-
     def test_ignore_replication_slot(self):
         self.assertFalse(self.c.ignore_replication_slot({'name': 'foo', 'type': 'physical',
                                                          'database': 'bar', 'plugin': 'wal2json'}))
diff --git a/tests/test_consul.py b/tests/test_consul.py
index 83ee67d8..494d1126 100644
--- a/tests/test_consul.py
+++ b/tests/test_consul.py
@@ -3,8 +3,10 @@ import unittest
 
 from consul import ConsulException, NotFound
 from mock import Mock, PropertyMock, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.consul import AbstractDCS, Cluster, Consul, ConsulInternalError, \
     ConsulError, ConsulClient, HTTPClient, InvalidSessionTTL, InvalidSession, RetryFailedError
+from patroni.postgresql.mpp import get_mpp
 from . import SleepException
 
 
@@ -91,13 +93,17 @@ class TestConsul(unittest.TestCase):
     @patch.object(consul.Consul.KV, 'get', kv_get)
     @patch.object(consul.Consul.KV, 'delete', Mock())
     def setUp(self):
-        Consul({'ttl': 30, 'scope': 't', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10,
-                'verify': 'on', 'key': 'foo', 'cert': 'bar', 'cacert': 'buz', 'token': 'asd', 'dc': 'dc1',
-                'register_service': True})
-        Consul({'ttl': 30, 'scope': 't_', 'name': 'p', 'url': 'https://l:1', 'retry_timeout': 10,
-                'verify': 'on', 'cert': 'bar', 'cacert': 'buz', 'register_service': True})
-        self.c = Consul({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'host': 'localhost:1', 'retry_timeout': 10,
-                         'register_service': True, 'service_check_tls_server_name': True})
+        self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't', 'name': 'p', 'retry_timeout': 10,
+                                       'consul': {'url': 'https://l:1', 'verify': 'on',
+                                                  'key': 'foo', 'cert': 'bar', 'cacert': 'buz',
+                                                  'token': 'asd', 'dc': 'dc1', 'register_service': True}}), Consul)
+        self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 't_', 'name': 'p', 'retry_timeout': 10,
+                                       'consul': {'url': 'https://l:1', 'verify': 'on',
+                                                  'cert': 'bar', 'cacert': 'buz', 'register_service': True}}), Consul)
+        self.c = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'postgresql1', 'retry_timeout': 10,
+                          'consul': {'host': 'localhost:1', 'register_service': True,
+                                     'service_check_tls_server_name': True}})
+        self.assertIsInstance(self.c, Consul)
         self.c._base_path = 'service/good'
         self.c.get_cluster()
 
@@ -130,7 +136,7 @@ class TestConsul(unittest.TestCase):
         self.assertIsInstance(self.c.get_cluster(), Cluster)
 
     def test__get_citus_cluster(self):
-        self.c._citus_group = '0'
+        self.c._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         cluster = self.c.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
diff --git a/tests/test_ctl.py b/tests/test_ctl.py
index f9ee62ce..f5341f88 100644
--- a/tests/test_ctl.py
+++ b/tests/test_ctl.py
@@ -12,6 +12,7 @@ from patroni.ctl import ctl, load_config, output_members, get_dcs, parse_dcs, \
     get_all_members, get_any_member, get_cursor, query_member, PatroniCtlException, apply_config_changes, \
     format_config_for_editing, show_diff, invoke_editor, format_pg_version, CONFIG_FILE_PATH, PatronictlPrettyTable
 from patroni.dcs import Cluster, Failover
+from patroni.postgresql.mpp import get_mpp
 from patroni.psycopg import OperationalError
 from patroni.utils import tzutc
 from prettytable import PrettyTable, ALL
@@ -69,7 +70,7 @@ class TestCtl(unittest.TestCase):
     @patch('patroni.psycopg.connect', psycopg_connect)
     def test_get_cursor(self):
         with click.Context(click.Command('query')) as ctx:
-            ctx.obj = {'__config': {}}
+            ctx.obj = {'__config': {}, '__mpp': get_mpp({})}
             for role in self.TEST_ROLES:
                 self.assertIsNone(get_cursor(get_cluster_initialized_without_leader(), None, {}, role=role))
                 self.assertIsNotNone(get_cursor(get_cluster_initialized_with_leader(), None, {}, role=role))
@@ -107,7 +108,7 @@ class TestCtl(unittest.TestCase):
 
     def test_output_members(self):
         with click.Context(click.Command('list')) as ctx:
-            ctx.obj = {'__config': {}}
+            ctx.obj = {'__config': {}, '__mpp': get_mpp({})}
             scheduled_at = datetime.now(tzutc) + timedelta(seconds=600)
             cluster = get_cluster_initialized_with_leader(Failover(1, 'foo', 'bar', scheduled_at))
             del cluster.members[1].data['conn_url']
@@ -250,7 +251,7 @@ class TestCtl(unittest.TestCase):
     @patch('patroni.dynamic_loader.iter_modules', Mock(return_value=['patroni.dcs.dummy', 'patroni.dcs.etcd']))
     def test_get_dcs(self):
         with click.Context(click.Command('list')) as ctx:
-            ctx.obj = {'__config': {'dummy': {}}}
+            ctx.obj = {'__config': {'dummy': {}}, '__mpp': get_mpp({})}
             self.assertRaises(PatroniCtlException, get_dcs, 'dummy', 0)
 
     @patch('patroni.psycopg.connect', psycopg_connect)
@@ -439,7 +440,7 @@ class TestCtl(unittest.TestCase):
 
     def test_get_any_member(self):
         with click.Context(click.Command('list')) as ctx:
-            ctx.obj = {'__config': {}}
+            ctx.obj = {'__config': {}, '__mpp': get_mpp({})}
             for role in self.TEST_ROLES:
                 self.assertIsNone(get_any_member(get_cluster_initialized_without_leader(), None, role=role))
 
@@ -448,7 +449,7 @@ class TestCtl(unittest.TestCase):
 
     def test_get_all_members(self):
         with click.Context(click.Command('list')) as ctx:
-            ctx.obj = {'__config': {}}
+            ctx.obj = {'__config': {}, '__mpp': get_mpp({})}
             for role in self.TEST_ROLES:
                 self.assertEqual(list(get_all_members(get_cluster_initialized_without_leader(), None, role=role)), [])
 
diff --git a/tests/test_etcd.py b/tests/test_etcd.py
index 874aac5c..d7a423be 100644
--- a/tests/test_etcd.py
+++ b/tests/test_etcd.py
@@ -5,8 +5,10 @@ import unittest
 
 from dns.exception import DNSException
 from mock import Mock, PropertyMock, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.etcd import AbstractDCS, EtcdClient, Cluster, Etcd, EtcdError, DnsCachingResolver
 from patroni.exceptions import DCSError
+from patroni.postgresql.mpp import get_mpp
 from patroni.utils import Retry
 from urllib3.exceptions import ReadTimeoutError
 
@@ -138,8 +140,9 @@ class TestClient(unittest.TestCase):
     @patch.object(EtcdClient, '_get_machines_list',
                   Mock(return_value=['http://localhost:2379', 'http://localhost:4001']))
     def setUp(self):
-        self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3,
-                          'srv': 'test', 'scope': 'test', 'name': 'foo'})
+        self.etcd = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 3,
+                             'etcd': {'srv': 'test'}, 'scope': 'test', 'name': 'foo'})
+        self.assertIsInstance(self.etcd, Etcd)
         self.client = self.etcd._client
         self.client.http.request = http_request
         self.client.http.request_encode_body = http_request
@@ -235,7 +238,7 @@ class TestEtcd(unittest.TestCase):
                   Mock(return_value=['http://localhost:2379', 'http://localhost:4001']))
     def setUp(self):
         self.etcd = Etcd({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
-                          'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'})
+                          'host': 'localhost:2379', 'scope': 'test', 'name': 'foo'}, get_mpp({}))
 
     def test_base_path(self):
         self.assertEqual(self.etcd._base_path, '/patroni/test')
@@ -270,7 +273,7 @@ class TestEtcd(unittest.TestCase):
         self.assertRaises(EtcdError, self.etcd.get_cluster)
 
     def test__get_citus_cluster(self):
-        self.etcd._citus_group = '0'
+        self.etcd._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         cluster = self.etcd.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
diff --git a/tests/test_etcd3.py b/tests/test_etcd3.py
index 10ab1ea5..fcfd4e4b 100644
--- a/tests/test_etcd3.py
+++ b/tests/test_etcd3.py
@@ -4,10 +4,12 @@ import unittest
 import urllib3
 
 from mock import Mock, PropertyMock, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.etcd import DnsCachingResolver
 from patroni.dcs.etcd3 import PatroniEtcd3Client, Cluster, Etcd3, Etcd3Client, \
     Etcd3Error, Etcd3ClientError, ReAuthenticateMode, RetryFailedError, InvalidAuthToken, Unavailable, \
     Unknown, UnsupportedEtcdVersion, UserEmpty, AuthFailed, AuthOldRevision, base64_encode
+from patroni.postgresql.mpp import get_mpp
 from threading import Thread
 
 from . import SleepException, MockResponse
@@ -80,9 +82,9 @@ class BaseTestEtcd3(unittest.TestCase):
     @patch.object(Thread, 'start', Mock())
     @patch.object(urllib3.PoolManager, 'urlopen', mock_urlopen)
     def setUp(self):
-        self.etcd3 = Etcd3({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10,
-                            'host': 'localhost:2378', 'scope': 'test', 'name': 'foo',
-                            'username': 'etcduser', 'password': 'etcdpassword'})
+        self.etcd3 = get_dcs({'namespace': '/patroni/', 'ttl': 30, 'retry_timeout': 10, 'name': 'foo', 'scope': 'test',
+                              'etcd3': {'host': 'localhost:2378', 'username': 'etcduser', 'password': 'etcdpassword'}})
+        self.assertIsInstance(self.etcd3, Etcd3)
         self.client = self.etcd3._client
         self.kv_cache = self.client._kv_cache
 
@@ -236,7 +238,7 @@ class TestEtcd3(BaseTestEtcd3):
             self.assertRaises(Etcd3Error, self.etcd3.get_cluster)
 
     def test__get_citus_cluster(self):
-        self.etcd3._citus_group = '0'
+        self.etcd3._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         cluster = self.etcd3.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
diff --git a/tests/test_exhibitor.py b/tests/test_exhibitor.py
index a908e1fc..5a72eb21 100644
--- a/tests/test_exhibitor.py
+++ b/tests/test_exhibitor.py
@@ -2,6 +2,7 @@ import unittest
 import urllib3
 
 from mock import Mock, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.exhibitor import ExhibitorEnsembleProvider, Exhibitor
 from patroni.dcs.zookeeper import ZooKeeperError
 
@@ -26,8 +27,9 @@ class TestExhibitor(unittest.TestCase):
         status=200, body=b'{"servers":["127.0.0.1","127.0.0.2","127.0.0.3"],"port":2181}')))
     @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})
+        self.e = get_dcs({'exhibitor': {'hosts': ['localhost', 'exhibitor'], 'port': 8181},
+                          'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10})
+        self.assertIsInstance(self.e, Exhibitor)
 
     @patch.object(ExhibitorEnsembleProvider, 'poll', Mock(return_value=True))
     @patch.object(MockKazooClient, 'get_children', Mock(side_effect=Exception))
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 5b1d4562..45f64164 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -197,7 +197,7 @@ def run_async(self, func, args=()):
 @patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=False))
 @patch('patroni.async_executor.AsyncExecutor.run_async', run_async)
 @patch('patroni.postgresql.rewind.Thread', Mock())
-@patch('patroni.postgresql.citus.CitusHandler.start', Mock())
+@patch('patroni.postgresql.mpp.citus.CitusHandler.start', Mock())
 @patch('subprocess.call', Mock(return_value=0))
 @patch('time.sleep', Mock())
 class TestHa(PostgresInit):
@@ -593,8 +593,8 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.bootstrap(), 'failed to acquire initialize lock')
 
     @patch('patroni.psycopg.connect', psycopg_connect)
-    @patch('patroni.postgresql.citus.connect', psycopg_connect)
-    @patch('patroni.postgresql.citus.quote_ident', Mock())
+    @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect)
+    @patch('patroni.postgresql.mpp.citus.quote_ident', Mock())
     @patch.object(Postgresql, 'connection', Mock(return_value=None))
     def test_bootstrap_initialized_new_cluster(self):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
@@ -615,8 +615,8 @@ class TestHa(PostgresInit):
         self.assertRaises(PatroniFatalException, self.ha.post_bootstrap)
 
     @patch('patroni.psycopg.connect', psycopg_connect)
-    @patch('patroni.postgresql.citus.connect', psycopg_connect)
-    @patch('patroni.postgresql.citus.quote_ident', Mock())
+    @patch('patroni.postgresql.mpp.citus.connect', psycopg_connect)
+    @patch('patroni.postgresql.mpp.citus.quote_ident', Mock())
     @patch.object(Postgresql, 'connection', Mock(return_value=None))
     def test_bootstrap_release_initialize_key_on_watchdog_failure(self):
         self.ha.cluster = get_cluster_not_initialized_without_leader()
@@ -659,7 +659,7 @@ class TestHa(PostgresInit):
     @patch.object(ConfigHandler, 'replace_pg_hba', Mock())
     @patch.object(ConfigHandler, 'replace_pg_ident', Mock())
     @patch.object(PostmasterProcess, 'start', Mock(return_value=MockPostmaster()))
-    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False))
     def test_worker_restart(self):
         self.ha.has_lock = true
         self.ha.patroni.request = Mock()
@@ -694,7 +694,7 @@ class TestHa(PostgresInit):
             self.ha.is_paused = true
             self.assertEqual(self.ha.run_cycle(), 'PAUSE: restart in progress')
 
-    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False))
     def test_manual_failover_from_leader(self):
         self.ha.has_lock = true  # I am the leader
 
@@ -733,7 +733,7 @@ class TestHa(PostgresInit):
                              ('Member %s exceeds maximum replication lag', 'b'))
             self.ha.cluster.members.pop()
 
-    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False))
     def test_manual_switchover_from_leader(self):
         self.ha.has_lock = true  # I am the leader
 
@@ -774,7 +774,7 @@ class TestHa(PostgresInit):
             self.assertEqual(self.ha.run_cycle(), 'no action. I am (postgresql0), the leader with the lock')
             self.assertEqual(mock_info.call_args_list[0][0], ('Member %s exceeds maximum replication lag', 'leader'))
 
-    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False))
     def test_scheduled_switchover_from_leader(self):
         self.ha.has_lock = true  # I am the leader
 
@@ -1544,7 +1544,7 @@ class TestHa(PostgresInit):
         self.ha.is_failover_possible = true
         self.ha.shutdown()
 
-    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False))
     def test_shutdown_citus_worker(self):
         self.ha.is_leader = true
         self.p.is_running = Mock(side_effect=[Mock(), False])
@@ -1656,7 +1656,7 @@ class TestHa(PostgresInit):
         self.assertRaises(DCSError, self.ha.acquire_lock)
         self.assertFalse(self.ha.acquire_lock())
 
-    @patch('patroni.postgresql.citus.CitusHandler.is_coordinator', Mock(return_value=False))
+    @patch('patroni.postgresql.mpp.AbstractMPPHandler.is_coordinator', Mock(return_value=False))
     def test_notify_citus_coordinator(self):
         self.ha.patroni.request = Mock()
         self.ha.notify_citus_coordinator('before_demote')
diff --git a/tests/test_kubernetes.py b/tests/test_kubernetes.py
index b6db7fb3..c493d799 100644
--- a/tests/test_kubernetes.py
+++ b/tests/test_kubernetes.py
@@ -8,9 +8,11 @@ import unittest
 import urllib3
 
 from mock import Mock, PropertyMock, mock_open, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.kubernetes import Cluster, k8s_client, k8s_config, K8sConfig, K8sConnectionFailed, \
     K8sException, K8sObject, Kubernetes, KubernetesError, KubernetesRetriableException, \
     Retry, RetryFailedError, SERVICE_HOST_ENV_NAME, SERVICE_PORT_ENV_NAME
+from patroni.postgresql.mpp import get_mpp
 from threading import Thread
 from . import MockResponse, SleepException
 
@@ -225,11 +227,12 @@ class BaseTestKubernetes(unittest.TestCase):
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_pod', mock_list_namespaced_pod, create=True)
     @patch.object(k8s_client.CoreV1Api, 'list_namespaced_config_map', mock_list_namespaced_config_map, create=True)
     def setUp(self, config=None):
-        config = config or {}
-        config.update(ttl=30, scope='test', name='p-0', loop_wait=10, group=0,
-                      retry_timeout=10, labels={'f': 'b'}, bypass_api_service=True)
-        self.k = Kubernetes(config)
-        self.k._citus_group = None
+        config = {'ttl': 30, 'scope': 'test', 'name': 'p-0', 'loop_wait': 10, 'retry_timeout': 10,
+                  'kubernetes': {'labels': {'f': 'b'}, 'bypass_api_service': True, **(config or {})},
+                  'citus': {'group': 0, 'database': 'postgres'}}
+        self.k = get_dcs(config)
+        self.assertIsInstance(self.k, Kubernetes)
+        self.k._mpp = get_mpp({})
         self.assertRaises(AttributeError, self.k._pods._build_cache)
         self.k._pods._is_ready = True
         self.assertRaises(TypeError, self.k._kinds._build_cache)
@@ -254,7 +257,7 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
             self.assertRaises(KubernetesError, self.k.get_cluster)
 
     def test__get_citus_cluster(self):
-        self.k._citus_group = '0'
+        self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         cluster = self.k.get_cluster()
         self.assertIsInstance(cluster, Cluster)
         self.assertIsInstance(cluster.workers[1], Cluster)
@@ -466,7 +469,7 @@ class TestCacheBuilder(BaseTestKubernetes):
     @patch('patroni.dcs.kubernetes.ObjectCache._watch', mock_watch)
     @patch.object(urllib3.HTTPResponse, 'read_chunked')
     def test__build_cache(self, mock_read_chunked):
-        self.k._citus_group = '0'
+        self.k._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         mock_read_chunked.return_value = [json.dumps(
             {'type': 'MODIFIED', 'object': {'metadata': {
                 'name': self.k.config_path, 'resourceVersion': '2', 'annotations': {self.k._CONFIG: 'foo'}}}}
diff --git a/tests/test_mpp.py b/tests/test_mpp.py
new file mode 100644
index 00000000..9eb87633
--- /dev/null
+++ b/tests/test_mpp.py
@@ -0,0 +1,52 @@
+from typing import Any
+from patroni.exceptions import PatroniException
+from patroni.postgresql.mpp import AbstractMPP, get_mpp, Null
+
+from . import BaseTestPostgresql
+from .test_ha import get_cluster_initialized_with_leader
+
+
+class TestMPP(BaseTestPostgresql):
+
+    def setUp(self):
+        super(TestMPP, self).setUp()
+        self.cluster = get_cluster_initialized_with_leader()
+
+    def test_get_handler_impl_exception(self):
+        class DummyMPP(AbstractMPP):
+            def __init__(self) -> None:
+                super().__init__({})
+
+            @staticmethod
+            def validate_config(config: Any) -> bool:
+                return True
+
+            @property
+            def group(self) -> None:
+                return None
+
+            @property
+            def coordinator_group_id(self) -> None:
+                return None
+
+            @property
+            def type(self) -> str:
+                return "dummy"
+
+        mpp = DummyMPP()
+        self.assertRaises(PatroniException, mpp.get_handler_impl, self.p)
+
+    def test_null_handler(self):
+        config = {}
+        mpp = get_mpp(config)
+        self.assertIsInstance(mpp, Null)
+        self.assertIsNone(mpp.group)
+        self.assertTrue(mpp.validate_config(config))
+        nullHandler = mpp.get_handler_impl(self.p)
+        self.assertIsNone(nullHandler.handle_event(self.cluster, {}))
+        self.assertIsNone(nullHandler.sync_meta_data(self.cluster))
+        self.assertIsNone(nullHandler.on_demote())
+        self.assertIsNone(nullHandler.schedule_cache_rebuild())
+        self.assertIsNone(nullHandler.bootstrap())
+        self.assertIsNone(nullHandler.adjust_postgres_gucs({}))
+        self.assertFalse(nullHandler.ignore_replication_slot({}))
diff --git a/tests/test_raft.py b/tests/test_raft.py
index 9bb109e9..387a5a56 100644
--- a/tests/test_raft.py
+++ b/tests/test_raft.py
@@ -4,8 +4,10 @@ import tempfile
 import time
 
 from mock import Mock, PropertyMock, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.raft import Cluster, DynMemberSyncObj, KVStoreTTL, \
     Raft, RaftError, SyncObjUtility, TCPTransport, _TCPTransport
+from patroni.postgresql.mpp import get_mpp
 from pysyncobj import SyncObjConf, FAIL_REASON
 
 
@@ -128,9 +130,10 @@ class TestRaft(unittest.TestCase):
     _TMP = tempfile.gettempdir()
 
     def test_raft(self):
-        raft = Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'self_addr': '127.0.0.1:1234',
-                     'retry_timeout': 10, 'data_dir': self._TMP,
-                     'database': 'citus', 'group': 0})
+        raft = get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'retry_timeout': 10,
+                        'raft': {'self_addr': '127.0.0.1:1234', 'data_dir': self._TMP},
+                        'citus': {'group': 0, 'database': 'postgres'}})
+        self.assertIsInstance(raft, Raft)
         raft.reload_config({'retry_timeout': 20, 'ttl': 60, 'loop_wait': 10})
         self.assertTrue(raft._sync_obj.set(raft.members_path + 'legacy', '{"version":"2.0.0"}'))
         self.assertTrue(raft.touch_member(''))
@@ -139,9 +142,9 @@ class TestRaft(unittest.TestCase):
         self.assertTrue(raft.set_config_value('{}'))
         self.assertTrue(raft.write_sync_state('foo', 'bar'))
         self.assertFalse(raft.write_sync_state('foo', 'bar', 1))
-        raft._citus_group = '1'
+        raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}})
         self.assertTrue(raft.manual_failover('foo', 'bar'))
-        raft._citus_group = '0'
+        raft._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         self.assertTrue(raft.take_leader())
         cluster = raft.get_cluster()
         self.assertIsInstance(cluster, Cluster)
@@ -157,9 +160,9 @@ class TestRaft(unittest.TestCase):
         self.assertTrue(raft.delete_sync_state())
         self.assertTrue(raft.set_history_value(''))
         self.assertTrue(raft.delete_cluster())
-        raft._citus_group = '1'
+        raft._mpp = get_mpp({'citus': {'group': 1, 'database': 'postgres'}})
         self.assertTrue(raft.delete_cluster())
-        raft._citus_group = None
+        raft._mpp = get_mpp({})
         raft.get_cluster()
         raft.watch(None, 0.001)
         raft._sync_obj.destroy()
@@ -175,5 +178,5 @@ class TestRaft(unittest.TestCase):
     def test_init(self, mock_event, mock_kvstore):
         mock_kvstore.return_value.applied_local_log = False
         mock_event.return_value.is_set.side_effect = [False, True]
-        self.assertIsNotNone(Raft({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True,
-                                   'self_addr': '1', 'data_dir': self._TMP}))
+        self.assertIsInstance(get_dcs({'ttl': 30, 'scope': 'test', 'name': 'pg', 'patronictl': True,
+                                       'raft': {'self_addr': '1', 'data_dir': self._TMP}}), Raft)
diff --git a/tests/test_zookeeper.py b/tests/test_zookeeper.py
index 3ce3ea75..3cd03467 100644
--- a/tests/test_zookeeper.py
+++ b/tests/test_zookeeper.py
@@ -7,8 +7,10 @@ from kazoo.handlers.threading import SequentialThreadingHandler
 from kazoo.protocol.states import KeeperState, WatchedEvent, ZnodeStat
 from kazoo.retry import RetryFailedError
 from mock import Mock, PropertyMock, patch
+from patroni.dcs import get_dcs
 from patroni.dcs.zookeeper import Cluster, PatroniKazooClient, \
     PatroniSequentialThreadingHandler, ZooKeeper, ZooKeeperError
+from patroni.postgresql.mpp import get_mpp
 
 
 class MockKazooClient(Mock):
@@ -148,9 +150,9 @@ class TestZooKeeper(unittest.TestCase):
 
     @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,
-                             'set_acls': {'CN=principal2': ['ALL']}})
+        self.zk = get_dcs({'scope': 'test', 'name': 'foo', 'ttl': 30, 'retry_timeout': 10, 'loop_wait': 10,
+                           'zookeeper': {'hosts': ['localhost:2181'], 'set_acls': {'CN=principal2': ['ALL']}}})
+        self.assertIsInstance(self.zk, ZooKeeper)
 
     def test_reload_config(self):
         self.zk.reload_config({'ttl': 20, 'retry_timeout': 10, 'loop_wait': 10})
@@ -177,7 +179,7 @@ class TestZooKeeper(unittest.TestCase):
         self.assertEqual(cluster.last_lsn, 500)
 
     def test__get_citus_cluster(self):
-        self.zk._citus_group = '0'
+        self.zk._mpp = get_mpp({'citus': {'group': 0, 'database': 'postgres'}})
         for _ in range(0, 2):
             cluster = self.zk.get_cluster()
             self.assertIsInstance(cluster, Cluster)

From dd548c49645c78d0b85cc723bf3bec500528c3da Mon Sep 17 00:00:00 2001
From: Alexander Kukushkin <cyberdemn@gmail.com>
Date: Thu, 21 Dec 2023 09:25:51 +0100
Subject: [PATCH 10/10] Create citus database and extension idempotently
 (#2990)

Consider a task: we want to create an extension _before_ citus in a database. Currently `post_bootstrab` script is executed before `CitusHandler.bootstrap()` method, which seems to allow doing that, but in fact `CitusHandler.bootstrap()` will fail to create already existing database and as a result the whole bootstrap will fail.

Changing the order of execution of `post_bootstrab` hook and `CitusHandler.bootstrap()` seems to be useless, because it will not allow creating another extension _before_ citus. Therefore the only way of solving it is making CREATE DATABASE and CREATE EXTENSION idempotent. It will allow to create citus database and all dependencies from the `post_bootstrab` hook.
---
 patroni/postgresql/mpp/citus.py | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/patroni/postgresql/mpp/citus.py b/patroni/postgresql/mpp/citus.py
index a17d1435..b8c205ce 100644
--- a/patroni/postgresql/mpp/citus.py
+++ b/patroni/postgresql/mpp/citus.py
@@ -8,7 +8,7 @@ from typing import Any, Dict, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from . import AbstractMPP, AbstractMPPHandler
 from ...dcs import Cluster
-from ...psycopg import connect, quote_ident
+from ...psycopg import connect, quote_ident, quote_literal
 from ...utils import parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -389,9 +389,16 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         if self._config['database'] != self._postgresql.database:
             conn = connect(**conn_kwargs)
             try:
+                database = self._config['database']
+                sql = """DO $$
+BEGIN
+    PERFORM * FROM pg_catalog.pg_database WHERE datname = {0};
+    IF NOT FOUND THEN
+        CREATE DATABASE {1};
+    END IF;
+END;$$""".format(quote_literal(database), quote_ident(database, conn))
                 with conn.cursor() as cur:
-                    cur.execute('CREATE DATABASE {0}'.format(
-                        quote_ident(self._config['database'], conn)).encode('utf-8'))
+                    cur.execute(sql.encode('utf-8'))
             finally:
                 conn.close()
 
@@ -399,7 +406,7 @@ class CitusHandler(Citus, AbstractMPPHandler, Thread):
         conn = connect(**conn_kwargs)
         try:
             with conn.cursor() as cur:
-                cur.execute('CREATE EXTENSION citus')
+                cur.execute('CREATE EXTENSION IF NOT EXISTS citus')
 
                 superuser = self._postgresql.config.superuser
                 params = {k: superuser[k] for k in ('password', 'sslcert', 'sslkey') if k in superuser}