config: don't apply values that fail validation

Previously, config_store::read_yaml() would apply property values even
when custom validation failed, only recording the error. This meant
invalid configuration values would silently take effect despite being
reported as errors.

This behavior was also inconsistent: conversion exceptions (e.g.,
YAML::BadConversion) correctly prevented values from being set, but
custom validator errors did not.

Fix by moving set_value() inside the else branch so values are only
applied when validation succeeds. This has two effects:

1. Optional properties now correctly retain their default value when
   validation fails.

2. Required properties now correctly throw std::invalid_argument when
   validation fails. Previously, the throw logic was bypassed because
   ok was set to true even after validation errors.

Note: It's unclear why we don't stop server startup on invalid optional
config values. To avoid breaking something I don't fully understand,
this change takes the conservative approach of retaining defaults.

Author: nvartolomei

src/v/config/config_store.h

@@ -94,9 +94,10 @@ class config_store {
                     errors[name] = fmt::format(
                       "Validation error: {}",
                       validation_err.value().error_message());
+                } else {
+                    prop->set_value(node.second);
+                    ok = true;
                 }
-                prop->set_value(node.second);
-                ok = true;
             } catch (const YAML::InvalidNode& e) {
                 errors[name] = fmt::format("Invalid syntax: {}", e);
             } catch (const YAML::ParserException& e) {

src/v/config/tests/config_store_test.cc

@@ -309,9 +309,7 @@ SEASTAR_THREAD_TEST_CASE(validate_with_validator_error) {
     BOOST_TEST(errors.size() > 0);
 
     // Property should retain default value
-    // Surprising but the current behavior is that invalid values actually
-    // update the property.
-    BOOST_TEST(cfg.validated_string() == "invalid_value");
+    BOOST_TEST(cfg.validated_string() == "magic_foo");
 }
 
 SEASTAR_THREAD_TEST_CASE(validate_with_type_mismatch) {
@@ -346,11 +344,7 @@ SEASTAR_THREAD_TEST_CASE(validate_required_with_validator_error) {
       "required_validated_string: invalid_value\n");
 
     // Required property with validation error throws std::invalid_argument
-    // BOOST_CHECK_THROW(cfg.read_yaml(invalid_yaml), std::invalid_argument);
-
-    auto errors = cfg.read_yaml(invalid_yaml);
-    BOOST_TEST(errors.size() > 0);
-    BOOST_TEST(cfg.required_validated_string() == "invalid_value");
+    BOOST_CHECK_THROW(cfg.read_yaml(invalid_yaml), std::invalid_argument);
 }
 
 SEASTAR_THREAD_TEST_CASE(config_json_serialization) {

tests/rptest/tests/adjacent_segment_merging_test.py

@@ -150,6 +150,10 @@ def __init__(self, test_context, *args, **kwargs):
             test_context, extra_rp_conf=xtra_conf, num_brokers=1, *args, **kwargs
         )
 
+        self.redpanda.set_environment(
+            {"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"}
+        )
+
     @cluster(num_nodes=2)
     @matrix(
         acks=[

tests/rptest/tests/e2e_shadow_indexing_test.py

@@ -1232,6 +1232,10 @@ def __init__(self, test_context):
         self.msg_size = 1024 * 256
         self.msg_count = 3000
 
+        self.redpanda.set_environment(
+            {"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"}
+        )
+
     def produce(self):
         topic_name = self.topics[0].name
         producer = KgoVerifierProducer(

tests/rptest/tests/nodes_decommissioning_test.py

@@ -70,6 +70,10 @@ def __init__(self, test_context):
             extra_rp_conf=extra_rp_conf,
         )
 
+        self.redpanda.set_environment(
+            {"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"}
+        )
+
     def setup(self):
         # defer starting redpanda to test body
         pass

tests/rptest/tests/offset_for_leader_epoch_archival_test.py

@@ -60,6 +60,10 @@ def __init__(self, test_context):
             si_settings=si_settings,
         )
 
+        self.redpanda.set_environment(
+            {"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"}
+        )
+
     def _alter_topic_retention_with_retry(self, topic):
         rpk = RpkTool(self.redpanda)
 

tests/rptest/tests/partition_movement_test.py

@@ -1025,7 +1025,10 @@ def test_shadow_indexing(
         throughput, records, moves, partitions = self._get_scale_params()
         install_opts = InstallOptions(install_previous_version=test_mixed_versions)
         self.start_redpanda(
-            num_nodes=3, install_opts=install_opts, extra_rp_conf=extra_rp_conf
+            num_nodes=3,
+            install_opts=install_opts,
+            extra_rp_conf=extra_rp_conf,
+            environment={"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"},
         )
 
         self.topic = "topic"
@@ -1094,7 +1097,10 @@ def test_cross_shard(
 
         install_opts = InstallOptions(install_previous_version=test_mixed_versions)
         self.start_redpanda(
-            num_nodes=3, install_opts=install_opts, extra_rp_conf=extra_rp_conf
+            num_nodes=3,
+            install_opts=install_opts,
+            extra_rp_conf=extra_rp_conf,
+            environment={"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"},
         )
 
         self.topic = "topic"

tests/rptest/tests/retention_policy_test.py

@@ -254,6 +254,10 @@ def __init__(self, test_context):
         self.rpk = RpkTool(self.redpanda)
         self.s3_bucket_name = si_settings.cloud_storage_bucket
 
+        self.redpanda.set_environment(
+            {"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"}
+        )
+
     def query_segments(self):
         return self.redpanda.node_storage(self.redpanda.nodes[0]).segments(
             "kafka", self.topic_name, 0
@@ -457,6 +461,10 @@ def __init__(self, test_context):
         self.rpk = RpkTool(self.redpanda)
         self.s3_bucket_name = si_settings.cloud_storage_bucket
 
+        self.redpanda.set_environment(
+            {"__REDPANDA_TEST_DISABLE_BOUNDED_PROPERTY_CHECKS": "ON"}
+        )
+
     @cluster(num_nodes=3)
     @matrix(cloud_storage_type=get_cloud_storage_type())
     def test_cloud_retention_deleted_segments_count(self, cloud_storage_type):