Ensure pg_replication_slot_advance() doesn't timeout (#2060)

The bigger gap between the slot flush LSN and the LSN we want to advance to becomes more time it takes for the call to finish.
Once started failing the "lag" will grow more or less infinitely, that have the following negative side-effects:
1. Size of pg_wal on the replica will grow
2. Since the hot_standby_feedback is forcefully enabled, the primary will stop cleaning up dead tuples
I.e., we are not only in danger of running out of disk space, but also increasing chances of transaction wraparound to happen.

In order to mitigate it, we want to set the `statement_timeout` to 0 before calling `pg_replication_slot_advance()`.

Since the call is happening from the main HA loop and could take more than `loop_wait`, the next heartbeat run could be delayed.
There is also a possibility that the call could take longer than `ttl` and the member key/session in DCS for a given replica expires, but, the slot LSN in DCS is updated by the primary every `loop_wait` seconds. Hence, we don't expect that the slot_advance() call will take significantly longer than the `loop_wait` and therefore chances of the member key/session to expire are very low.

patroni/postgresql/slots.py

@@ -143,9 +143,9 @@ class SlotsHandler(object):
                 self._schedule_load_slots = True
 
     @contextmanager
-    def _get_local_connection_cursor(self, database):
+    def _get_local_connection_cursor(self, **kwargs):
         conn_kwargs = self._postgresql.config.local_connect_kwargs
-        conn_kwargs['database'] = database
+        conn_kwargs.update(kwargs)
         with get_connection_cursor(**conn_kwargs) as cur:
             yield cur
 
@@ -162,7 +162,7 @@ class SlotsHandler(object):
 
         # Create new logical slots
         for database, values in logical_slots.items():
-            with self._get_local_connection_cursor(database) as cur:
+            with self._get_local_connection_cursor(database=database) as cur:
                 for name, value in values.items():
                     try:
                         cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)" +
@@ -194,7 +194,7 @@ class SlotsHandler(object):
 
         # Advance logical slots
         for database, values in advance_slots.items():
-            with self._get_local_connection_cursor(database) as cur:
+            with self._get_local_connection_cursor(database=database, options='-c statement_timeout=0') as cur:
                 for name, value in values.items():
                     try:
                         cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",