config: don't apply values that fail validation

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

@@ -13,21 +13,33 @@
 #include "json/writer.h"
 
 #include <seastar/core/sstring.hh>
-#include <seastar/core/thread.hh>
 #include <seastar/testing/thread_test_case.hh>
 #include <seastar/util/log.hh>
 
 #include <cstdint>
 #include <iostream>
 #include <iterator>
-#include <random>
 #include <string>
 #include <utility>
 
 namespace {
 
 ss::logger lg("config_test"); // NOLINT
 
+std::optional<ss::sstring> validate_magic_prefix(const ss::sstring& value) {
+    if (!value.starts_with("magic_")) {
+        return fmt::format("value must start with 'magic_', got: {}", value);
+    }
+    return std::nullopt;
+}
+
+std::optional<ss::sstring> validate_throwing(const ss::sstring& value) {
+    if (value == "throw") {
+        throw std::runtime_error("validator threw an exception");
+    }
+    return std::nullopt;
+}
+
 struct test_config : public config::config_store {
     config::property<int> optional_int;
     config::property<ss::sstring> required_string;
@@ -44,6 +56,8 @@ struct test_config : public config::config_store {
     config::property<ss::sstring> default_secret_string;
     config::property<ss::sstring> secret_string;
     config::property<bool> aliased_bool;
+    config::property<ss::sstring> validated_string;
+    config::property<ss::sstring> throwing_validator_string;
 
     test_config()
       : optional_int(
@@ -113,11 +127,38 @@ struct test_config : public config::config_store {
           "aliased_bool",
           "Property with a compat alias",
           {.aliases = {"aliased_bool_legacy"}},
-          true) {}
+          true)
+      , validated_string(
+          *this,
+          "validated_string",
+          "String that must start with magic_",
+          {},
+          "magic_foo",
+          &validate_magic_prefix)
+      , throwing_validator_string(
+          *this,
+          "throwing_validator_string",
+          "String with a validator that throws",
+          {},
+          "safe",
+          &validate_throwing) {}
 };
 
 struct noop_config : public config::config_store {};
 
+struct required_validated_config : public config::config_store {
+    config::property<ss::sstring> required_validated_string;
+
+    required_validated_config()
+      : required_validated_string(
+          *this,
+          "required_validated_string",
+          "Required string that must start with magic_",
+          {.required = config::required::yes},
+          "magic_bar",
+          &validate_magic_prefix) {}
+};
+
 YAML::Node minimal_valid_configuration() {
     return YAML::Load(
       "required_string: test_value_1\n"
@@ -257,10 +298,53 @@ SEASTAR_THREAD_TEST_CASE(validate_valid_configuration) {
     BOOST_TEST(errors.size() == 0);
 }
 
-SEASTAR_THREAD_TEST_CASE(validate_invalid_configuration) {
+SEASTAR_THREAD_TEST_CASE(validate_with_validator_error) {
     auto cfg = test_config();
-    auto errors = cfg.read_yaml(valid_configuration());
-    BOOST_TEST(errors.size() == 0);
+
+    auto invalid_yaml = YAML::Load("validated_string: invalid_value\n");
+
+    auto errors = cfg.read_yaml(invalid_yaml);
+
+    // Should have error from validator
+    BOOST_TEST(errors.size() > 0);
+
+    // Property should retain default value
+    BOOST_TEST(cfg.validated_string() == "magic_foo");
+}
+
+SEASTAR_THREAD_TEST_CASE(validate_with_type_mismatch) {
+    auto cfg = test_config();
+
+    // Provide string where int is expected
+    auto invalid_yaml = YAML::Load("optional_int: not_an_int\n");
+
+    auto errors = cfg.read_yaml(invalid_yaml);
+
+    BOOST_REQUIRE(!errors.empty());
+    auto& [key, msg] = *errors.begin();
+    BOOST_TEST(key == "optional_int");
+    BOOST_TEST_INFO(msg);
+    BOOST_TEST(msg.contains("bad conversion"));
+
+    BOOST_TEST(cfg.optional_int() == 100); // expect default retained
+}
+
+SEASTAR_THREAD_TEST_CASE(validate_with_throwing_validator) {
+    auto cfg = test_config();
+
+    auto yaml = YAML::Load("throwing_validator_string: throw\n");
+
+    BOOST_CHECK_THROW(cfg.read_yaml(yaml), std::runtime_error);
+}
+
+SEASTAR_THREAD_TEST_CASE(validate_required_with_validator_error) {
+    auto cfg = required_validated_config();
+
+    auto invalid_yaml = YAML::Load(
+      "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);
 }
 
 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/cluster_config_test.py

@@ -205,6 +205,38 @@ def test_unknown_redpanda_yaml(self):
         )
 
 
+class ClusterConfigBoundedPropertyTest(RedpandaTest):
+    """
+    Test that bounded properties clamp out-of-range values from bootstrap config.
+    """
+
+    def __init__(self, *args, **kwargs):
+        # Set log_segment_size to 100 bytes (below minimum of 1MB)
+        super().__init__(
+            *args,
+            extra_rp_conf={"log_segment_size": 100},
+            **kwargs,
+        )
+
+    @cluster(num_nodes=1)
+    def test_bounded_property_clamped_to_minimum(self):
+        """
+        Verify that bounded properties are clamped to their minimum bound when
+        an out-of-range value is provided in bootstrap config. Setting
+        log_segment_size to 100 should result in the value being clamped to
+        1MB (the minimum bound).
+        """
+        admin = Admin(self.redpanda)
+        config = admin.get_cluster_config()
+
+        # The value should be clamped to 1MB (1048576 bytes)
+        min_segment_size = 1024 * 1024  # 1 MiB
+        assert config["log_segment_size"] == min_segment_size, (
+            f"Expected log_segment_size to be clamped to {min_segment_size} (1MB), "
+            f"but got {config['log_segment_size']}"
+        )
+
+
 class HasRedpandaAndAdmin(Protocol):
     redpanda: RedpandaService
     admin: Admin

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):