Merge pull request #15874 from andijcr/fix/v23.2.x/cfg_unchecked_test

[v23.2.x] Fix/v23.2.x/cfg unchecked test

src/v/cluster/config_manager.cc

@@ -345,8 +345,8 @@ static void preload_local(
   std::optional<std::reference_wrapper<config_manager::preload_result>>
     result) {
     auto& cfg = config::shard_local_cfg();
-    auto& property = cfg.get(key);
     if (cfg.contains(key)) {
+        auto& property = cfg.get(key);
         std::string raw_value;
         try {
             raw_value = value.as<std::string>();

tests/rptest/services/admin.py

@@ -379,15 +379,11 @@ def get_cluster_config_schema(self, node=None):
         return self._request("GET", "cluster_config/schema", node=node).json()
 
     def patch_cluster_config(self,
-                             upsert=None,
-                             remove=None,
+                             upsert: dict[str, str] = {},
+                             remove: list[str] = [],
                              force=False,
                              dry_run=False,
                              node=None):
-        if upsert is None:
-            upsert = {}
-        if remove is None:
-            remove = []
 
         path = "cluster_config"
         params = {}
@@ -844,7 +840,7 @@ def _get_details():
 
     def maintenance_start(self, node, dst_node=None):
         """
-        Start maintenanceing on 'node', sending the request to 'dst_node'.
+        Start maintenance on 'node', sending the request to 'dst_node'.
         """
         id = self.redpanda.node_id(node)
         url = f"brokers/{id}/maintenance"
@@ -854,7 +850,7 @@ def maintenance_start(self, node, dst_node=None):
 
     def maintenance_stop(self, node, dst_node=None):
         """
-        Stop maintenanceing on 'node', sending the request to 'dst_node'.
+        Stop maintenance on 'node', sending the request to 'dst_node'.
         """
         id = self.redpanda.node_id(node)
         url = f"brokers/{id}/maintenance"

tests/rptest/tests/cluster_config_test.py

@@ -1840,3 +1840,29 @@ def test_legacy_default_explicit_after_upgrade(self, wipe_cache: bool):
         self.redpanda.set_cluster_config({self.key: expected})
 
         self._check_value_everywhere(self.key, expected)
+
+
+class ClusterConfigUnknownTest(RedpandaTest):
+    def __init__(self, test_context):
+        super().__init__(test_context)
+
+        self.admin = Admin(self.redpanda)
+
+    @cluster(num_nodes=3)
+    def test_unknown_value(self):
+        """
+        Test that an unknown property saved in the log does not prevent a node from starting.
+        In this test we use a non existing property,
+        but it can be a property not yet introduced in the current version.
+        For example a property set in the middle of an upgrade followed up by a rollback.
+        see issues/15839
+        """
+        self.admin.patch_cluster_config(
+            upsert={"a_non_existing_property": "a_value_with_no_importance"},
+            force=True)
+
+        assert "a_non_existing_property" not in self.admin.get_cluster_config(
+        ), "unexpected property found in cluster config"
+
+        # issue would appear when reloading the property back
+        self.redpanda.restart_nodes(self.redpanda.nodes[0])