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
        if config:
            self._connection = postgresql.connection_pool.get(
                'citus', {'dbname': config['database'],
                          'options': '-c statement_timeout=0 -c idle_in_transaction_session_timeout=0'})
        self._pg_dist_group: Dict[int, PgDistTask] = {}  # Cache of pg_dist_node: {groupid: PgDistTask()}
        self._tasks: List[PgDistTask] = []  # Requests to change pg_dist_group, every task is a `PgDistTask`
        self._in_flight: Optional[PgDistTask] = None  # Reference to the `PgDistTask` being changed in a transaction
        self._schedule_load_pg_dist_group = True  # Flag that "pg_dist_group" should be queried from the database
        self._condition = Condition()  # protects _pg_dist_group, _tasks, _in_flight, and _schedule_load_pg_dist_group
        self.schedule_cache_rebuild()
"Initialize a new instance of :class:`CitusHandler`.

        :param postgresql: the Postgres node.
        :param config: the ``citus`` MPP config section.
        
patroni/postgresql/mpp/citus.py