Repository: zalando/patroni
Pull request #1926 (2021-04-29 14:59:33)

Compatibility with PostgreSQL 14
PostgreSQL 14 changed the behavior of replicas when certain parameters (like for example `max_connections`) are changed (increased): https://github.com/postgres/postgres/commit/15251c0a.
Instead of immediately exiting Postgres 14 pauses replication and waits for actions from the operator.

Since the `pg_is_wal_replay_paused()` returning `True` is the only indicator of such a change, Patroni on the replica will call the
`pg_wal_replay_resume()`, which would cause either continue replication or shutdown (like previously).

So far Patroni was never calling `pg_wal_replay_resume()` on its own, therefore, to remain backward compatible it will call it only for PostgreSQL 14+.

Jan-M: :+1:
CyberDem0n: :+1:

Patch:
diff --git a/.github/workflows/install_deps.py b/.github/workflows/install_deps.py
index fb5194cc..187d634f 100644
--- a/.github/workflows/install_deps.py
+++ b/.github/workflows/install_deps.py
@@ -33,13 +33,17 @@ def install_requirements(what):
 
 
 def install_packages(what):
+    from mapping import versions
+
     packages = {
         'zookeeper': ['zookeeper', 'zookeeper-bin', 'zookeeperd'],
         'consul': ['consul'],
     }
     packages['exhibitor'] = packages['zookeeper']
     packages = packages.get(what, [])
-    ver = str({'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what))
+    ver = versions.get(what)
+    subprocess.call(['sudo', 'sed', '-i', 's/pgdg main.*$/pgdg main {0}/'.format(ver),
+                     '/etc/apt/sources.list.d/pgdg.list'])
     subprocess.call(['sudo', 'apt-get', 'update', '-y'])
     return subprocess.call(['sudo', 'apt-get', 'install', '-y', 'postgresql-' + ver, 'expect-dev', 'wget'] + packages)
 
diff --git a/.github/workflows/mapping.py b/.github/workflows/mapping.py
new file mode 100644
index 00000000..abe30a39
--- /dev/null
+++ b/.github/workflows/mapping.py
@@ -0,0 +1,1 @@
+versions = {'etcd': '9.6', 'etcd3': '14', 'consul': '13', 'exhibitor': '12', 'raft': '11', 'kubernetes': '14'}
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 30891889..3aaeada6 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -29,7 +29,7 @@ Feature: basic replication
     Then I receive a response code 200
 
   Scenario: check stuck sync replica
-    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": true, "maximum_lag_on_syncnode": 15000000, "postgresql": {"parameters": {"synchronous_commit": "remote_apply"}}}
     Then I receive a response code 200
     And I create table on postgres0
     And table mytest is present on postgres1 after 2 seconds
@@ -43,7 +43,7 @@ Feature: basic replication
     Then I receive a response code 200
     When I issue a GET request to http://127.0.0.1:8010/async
     Then I receive a response code 200
-    When I issue a PATCH request to http://127.0.0.1:8008/config with {"maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"pause": null, "maximum_lag_on_syncnode": -1, "postgresql": {"parameters": {"synchronous_commit": "on"}}}
     Then I receive a response code 200
     And I drop table on postgres0
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 507d586b..25935564 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -417,6 +417,9 @@ class Ha(object):
         if msg:
             return msg
 
+        if not self.is_paused():
+            self.state_handler.handle_parameter_change()
+
         role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica'
         # It might happen that leader key in the standby cluster references non-exiting member.
         # In this case it is safe to continue running without changing recovery.conf
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index b33fff8b..2b0a7c2a 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -367,6 +367,17 @@ class Postgresql(object):
     def is_leader(self):
         return bool(self._cluster_info_state_get('timeline'))
 
+    def replay_paused(self):
+        return self._cluster_info_state_get('replay_paused')
+
+    def resume_wal_replay(self):
+        self._query('SELECT pg_catalog.pg_{0}_replay_resume()'.format(self.wal_name))
+
+    def handle_parameter_change(self):
+        if self.major_version >= 140000 and self.replay_paused():
+            logger.info('Resuming paused WAL replay for PostgreSQL 14+')
+            self.resume_wal_replay()
+
     def pg_control_timeline(self):
         try:
             return int(self.controldata().get("Latest checkpoint's TimeLineID"))
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index 519b0b24..4f40f189 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -189,7 +189,9 @@ class Rewind(object):
                     need_rewind = False
                 elif master_timeline > 1:
                     cur.execute('TIMELINE_HISTORY %s', (master_timeline,))
-                    history = bytes(cur.fetchone()[1]).decode('utf-8')
+                    history = cur.fetchone()[1]
+                    if not isinstance(history, six.string_types):
+                        history = bytes(history).decode('utf-8')
                     logger.debug('master: history=%s', history)
         except Exception:
             return logger.exception('Exception when working with master via replication connection')
diff --git a/patroni/postgresql/validator.py b/patroni/postgresql/validator.py
index 93cdce54..945d5afd 100644
--- a/patroni/postgresql/validator.py
+++ b/patroni/postgresql/validator.py
@@ -108,7 +108,7 @@ parameters = CaseInsensitiveDict({
     ),
     'archive_timeout': Integer(90300, None, 0, 1073741823, 's'),
     'array_nulls': Bool(90300, None),
-    'authentication_timeout': Integer(90300, None,  1, 600, 's'),
+    'authentication_timeout': Integer(90300, None, 1, 600, 's'),
     'autovacuum': Bool(90300, None),
     'autovacuum_analyze_scale_factor': Real(90300, None, 0, 100, None),
     'autovacuum_analyze_threshold': Integer(90300, None, 0, 2147483647, None),
@@ -151,12 +151,14 @@ parameters = CaseInsensitiveDict({
         Integer(90600, None, 30, 86400, 's')
     ),
     'checkpoint_warning': Integer(90300, None, 0, 2147483647, 's'),
+    'client_connection_check_interval': Integer(140000, None, '0', '2147483647', 'ms'),
     'client_encoding': String(90300, None),
     'client_min_messages': Enum(90300, None, ('debug5', 'debug4', 'debug3', 'debug2',
                                               'debug1', 'log', 'notice', 'warning', 'error')),
     'cluster_name': String(90500, None),
     'commit_delay': Integer(90300, None, 0, 100000, None),
     'commit_siblings': Integer(90300, None, 0, 1000, None),
+    'compute_query_id': EnumBool(140000, None, ('auto',)),
     'config_file': String(90300, None),
     'constraint_exclusion': EnumBool(90300, None, ('partition',)),
     'cpu_index_tuple_cost': Real(90300, None, 0, 1.79769e+308, None),
@@ -168,6 +170,7 @@ parameters = CaseInsensitiveDict({
     'DateStyle': String(90300, None),
     'db_user_namespace': Bool(90300, None),
     'deadlock_timeout': Integer(90300, None, 1, 2147483647, 'ms'),
+    'debug_invalidate_system_caches_always': Integer(140000, None, '0', '0', None),
     'debug_pretty_print': Bool(90300, None),
     'debug_print_parse': Bool(90300, None),
     'debug_print_plan': Bool(90300, None),
@@ -176,6 +179,7 @@ parameters = CaseInsensitiveDict({
     'default_table_access_method': String(120000, None),
     'default_tablespace': String(90300, None),
     'default_text_search_config': String(90300, None),
+    'default_toast_compression': Enum(140000, None, ('pglz', 'lz4')),
     'default_transaction_deferrable': Bool(90300, None),
     'default_transaction_isolation': Enum(90300, None, ('serializable', 'repeatable read',
                                                         'read committed', 'read uncommitted')),
@@ -188,6 +192,7 @@ parameters = CaseInsensitiveDict({
     ),
     'effective_cache_size': Integer(90300, None, 1, 2147483647, '8kB'),
     'effective_io_concurrency': Integer(90300, None, 0, 1000, None),
+    'enable_async_append': Bool(140000, None),
     'enable_bitmapscan': Bool(90300, None),
     'enable_gathermerge': Bool(100000, None),
     'enable_hashagg': Bool(90300, None),
@@ -203,12 +208,14 @@ parameters = CaseInsensitiveDict({
     'enable_partition_pruning': Bool(110000, None),
     'enable_partitionwise_aggregate': Bool(110000, None),
     'enable_partitionwise_join': Bool(110000, None),
+    'enable_resultcache': Bool(140000, None),
     'enable_seqscan': Bool(90300, None),
     'enable_sort': Bool(90300, None),
     'enable_tidscan': Bool(90300, None),
     'escape_string_warning': Bool(90300, None),
     'event_source': String(90300, None),
     'exit_on_error': Bool(90300, None),
+    'extension_destdir': String(140000, None),
     'external_pid_file': String(90300, None),
     'extra_float_digits': Integer(90300, None, -15, 3, None),
     'force_parallel_mode': EnumBool(90600, None, ('regress',)),
@@ -229,8 +236,10 @@ parameters = CaseInsensitiveDict({
     'hot_standby': Bool(90300, None),
     'hot_standby_feedback': Bool(90300, None),
     'huge_pages': EnumBool(90400, None, ('try',)),
+    'huge_page_size': Integer(140000, None, '0', '2147483647', 'kB'),
     'ident_file': String(90300, None),
     'idle_in_transaction_session_timeout': Integer(90600, None, 0, 2147483647, 'ms'),
+    'idle_session_timeout': Integer(140000, None, '0', '2147483647', 'ms'),
     'ignore_checksum_failure': Bool(90300, None),
     'ignore_invalid_pages': Bool(130000, None),
     'ignore_system_indexes': Bool(90300, None),
@@ -283,6 +292,7 @@ parameters = CaseInsensitiveDict({
     'log_parameter_max_length_on_error': Integer(130000, None, -1, 1073741823, 'B'),
     'log_parser_stats': Bool(90300, None),
     'log_planner_stats': Bool(90300, None),
+    'log_recovery_conflict_waits': Bool(140000, None),
     'log_replication_commands': Bool(90500, None),
     'log_rotation_age': Integer(90300, None, 0, 35791394, 'min'),
     'log_rotation_size': Integer(90300, None, 0, 2097151, 'kB'),
@@ -336,6 +346,7 @@ parameters = CaseInsensitiveDict({
         Integer(90400, 90600, 1, 8388607, None),
         Integer(90600, None, 0, 262143, None)
     ),
+    'min_dynamic_shared_memory': Integer(140000, None, '0', '2147483647', 'MB'),
     'min_parallel_index_scan_size': Integer(100000, None, 0, 715827882, '8kB'),
     'min_parallel_relation_size': Integer(90600, 100000, 0, 715827882, '8kB'),
     'min_parallel_table_scan_size': Integer(100000, None, 0, 715827882, '8kB'),
@@ -344,7 +355,7 @@ parameters = CaseInsensitiveDict({
         Integer(100000, None, 2, 2147483647, 'MB')
     ),
     'old_snapshot_threshold': Integer(90600, None, -1, 86400, 'min'),
-    'operator_precedence_warning': Bool(90500, None),
+    'operator_precedence_warning': Bool(90500, 140000),
     'parallel_leader_participation': Bool(110000, None),
     'parallel_setup_cost': Real(90600, None, 0, 1.79769e+308, None),
     'parallel_tuple_cost': Real(90600, None, 0, 1.79769e+308, None),
@@ -358,6 +369,8 @@ parameters = CaseInsensitiveDict({
     'pre_auth_delay': Integer(90300, None, 0, 60, 's'),
     'quote_all_identifiers': Bool(90300, None),
     'random_page_cost': Real(90300, None, 0, 1.79769e+308, None),
+    'recovery_init_sync_method': Enum(140000, None, ('fsync', 'syncfs')),
+    'remove_temp_files_after_crash': Bool(140000, None),
     'replacement_sort_tuples': Integer(90600, 110000, 0, 2147483647, None),
     'restart_after_crash': Bool(90300, None),
     'row_security': Bool(90500, None),
@@ -373,6 +386,7 @@ parameters = CaseInsensitiveDict({
     'ssl_ca_file': String(90300, None),
     'ssl_cert_file': String(90300, None),
     'ssl_ciphers': String(90300, None),
+    'ssl_crl_dir': String(140000, None),
     'ssl_crl_file': String(90300, None),
     'ssl_dh_params_file': String(100000, None),
     'ssl_ecdh_curve': String(90400, None),
@@ -388,7 +402,7 @@ parameters = CaseInsensitiveDict({
     'stats_temp_directory': String(90300, None),
     'superuser_reserved_connections': (
         Integer(90300, 90600, 0, 8388607, None),
-        Integer(90600, None, 0, 262143, None),
+        Integer(90600, None, 0, 262143, None)
     ),
     'synchronize_seqscans': Bool(90300, None),
     'synchronous_commit': (
@@ -424,6 +438,7 @@ parameters = CaseInsensitiveDict({
     'track_counts': Bool(90300, None),
     'track_functions': Enum(90300, None, ('none', 'pl', 'all')),
     'track_io_timing': Bool(90300, None),
+    'track_wal_io_timing': Bool(140000, None),
     'transaction_deferrable': Bool(90300, None),
     'transaction_isolation': Enum(90300, None, ('serializable', 'repeatable read',
                                                 'read committed', 'read uncommitted')),
@@ -433,7 +448,7 @@ parameters = CaseInsensitiveDict({
     'unix_socket_group': String(90300, None),
     'unix_socket_permissions': Integer(90300, None, 0, 511, None),
     'update_process_title': Bool(90300, None),
-    'vacuum_cleanup_index_scale_factor': Real(110000, None, 0, 1e+10, None),
+    'vacuum_cleanup_index_scale_factor': Real(110000, 140000, 0, 1e+10, None),
     'vacuum_cost_delay': (
         Integer(90300, 120000, 0, 100, 'ms'),
         Real(120000, None, 0, 100, 'ms')
@@ -443,8 +458,10 @@ parameters = CaseInsensitiveDict({
     'vacuum_cost_page_hit': Integer(90300, None, 0, 10000, None),
     'vacuum_cost_page_miss': Integer(90300, None, 0, 10000, None),
     'vacuum_defer_cleanup_age': Integer(90300, None, 0, 1000000, None),
+    'vacuum_failsafe_age': Integer(140000, None, '0', '2100000000', None),
     'vacuum_freeze_min_age': Integer(90300, None, 0, 1000000000, None),
     'vacuum_freeze_table_age': Integer(90300, None, 0, 2000000000, None),
+    'vacuum_multixact_failsafe_age': Integer(140000, None, '0', '2100000000', None),
     'vacuum_multixact_freeze_min_age': Integer(90300, None, 0, 1000000000, None),
     'vacuum_multixact_freeze_table_age': Integer(90300, None, 0, 2000000000, None),
     'wal_buffers': Integer(90300, None, -1, 262143, '8kB'),


Test patch:
diff --git a/.github/workflows/run_tests.py b/.github/workflows/run_tests.py
index fbda3999..89a9b222 100644
--- a/.github/workflows/run_tests.py
+++ b/.github/workflows/run_tests.py
@@ -23,7 +23,9 @@ def main():
 
     env = os.environ.copy()
     if sys.platform.startswith('linux'):
-        version = {'etcd': '9.6', 'etcd3': '13', 'consul': 12, 'exhibitor': 11, 'kubernetes': 13, 'raft': 12}.get(what)
+        from mapping import versions
+
+        version = versions.get(what)
         path = '/usr/lib/postgresql/{0}/bin:.'.format(version)
         unbuffer = ['timeout', '600', 'unbuffer']
         args = ['--tags=-skip'] if what == 'etcd' else []
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 42d01e6b..0fbc1b4d 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -725,3 +725,8 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'is_running', Mock(return_value=True))
     def test_set_enforce_hot_standby_feedback(self):
         self.p.set_enforce_hot_standby_feedback(True)
+
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=140000))
+    @patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value=True))
+    def test_handle_parameter_change(self):
+        self.p.handle_parameter_change()