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)