config: add "legacy defaults" capability and decrease partitions per shard default

src/v/config/base_property.h

@@ -14,6 +14,7 @@
 #include "json/stringbuffer.h"
 #include "json/writer.h"
 #include "seastarx.h"
+#include "utils/named_type.h"
 
 #include <seastar/util/bool_class.hh>
 
@@ -56,6 +57,10 @@ enum class odd_even_constraint {
     odd,
 };
 
+// This is equivalent to cluster::cluster_version, but defined here to
+// avoid a dependency between config/ and cluster/
+using legacy_version = named_type<int64_t, struct legacy_version_tag>;
+
 std::string_view to_string_view(visibility v);
 
 class base_property {
@@ -126,6 +131,12 @@ class base_property {
     virtual base_property& operator=(const base_property&) = 0;
     virtual ~base_property() noexcept = default;
 
+    /**
+     * Notify the property of the cluster's original logical version, in case
+     * it has alternative defaults for old clusters.
+     */
+    virtual void notify_original_version(legacy_version) = 0;
+
 private:
     friend std::ostream& operator<<(std::ostream&, const base_property&);
     std::string_view _name;

src/v/config/bounded_property.h

@@ -120,7 +120,8 @@ class bounded_property : public property<T> {
       std::string_view desc,
       base_property::metadata meta,
       T def,
-      numeric_bounds<I> bounds)
+      numeric_bounds<I> bounds,
+      std::optional<legacy_default<T>> legacy = std::nullopt)
       : property<T>(
         conf,
         name,
@@ -141,7 +142,8 @@ class bounded_property : public property<T> {
             } else {
                 return _bounds.validate(new_value);
             }
-        })
+        },
+        legacy)
       , _bounds(bounds)
       , _example(generate_example()) {}
 

src/v/config/config_store.h

@@ -153,6 +153,12 @@ class config_store {
         return o;
     }
 
+    void notify_original_version(legacy_version ov) {
+        for (const auto& [name, property] : _properties) {
+            property->notify_original_version(ov);
+        }
+    }
+
     virtual ~config_store() noexcept = default;
 
 private:

src/v/config/configuration.cc

@@ -188,15 +188,16 @@ configuration::configuration()
       "Maximum number of partitions which may be allocated to one shard (CPU "
       "core)",
       {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
-      7000,
+      1000,
       {
         .min = 16,    // Forbid absurdly small values that would prevent most
                       // practical workloads from running
         .max = 131072 // An upper bound to prevent pathological values, although
                       // systems will most likely hit issues far before reaching
                       // this.  This property is principally intended to be
                       // tuned downward from the default, not upward.
-      })
+      },
+      legacy_default<uint32_t>(7000, legacy_version{9}))
   , topic_partitions_reserve_shard0(
       *this,
       "topic_partitions_reserve_shard0",

src/v/config/property.h

@@ -36,6 +36,29 @@ class binding;
 template<typename T>
 class mock_property;
 
+/**
+ * An alternative default that only applies to clusters created before a
+ * particular logical version.
+ *
+ * This enables changing defaults for new clusters without disrupting
+ * existing clusters.
+ *
+ * **Be aware** that this only works for properties that are read _after_
+ * the bootstrap phase of startup.
+ */
+template<class T>
+class legacy_default {
+public:
+    legacy_default() = delete;
+
+    legacy_default(T v, legacy_version ov)
+      : value(std::move(v))
+      , max_original_version(ov) {}
+
+    T value;
+    legacy_version max_original_version;
+};
+
 template<class T>
 class property : public base_property {
 public:
@@ -48,10 +71,12 @@ class property : public base_property {
       std::string_view desc,
       base_property::metadata meta = {},
       T def = T{},
-      property::validator validator = property::noop_validator)
+      property::validator validator = property::noop_validator,
+      std::optional<legacy_default<T>> ld = std::nullopt)
       : base_property(conf, name, desc, meta)
       , _value(def)
       , _default(std::move(def))
+      , _legacy_default(std::move(ld))
       , _validator(std::move(validator)) {}
 
     /**
@@ -178,28 +203,48 @@ class property : public base_property {
         }
     }
 
+    void notify_original_version(legacy_version ov) override {
+        if (!_legacy_default.has_value()) {
+            // Most properties have no legacy default, and ignore this.
+            return;
+        }
+
+        if (ov < _legacy_default.value().max_original_version) {
+            _default = _legacy_default.value().value;
+            // In case someone already made a binding to us early in startup
+            notify_watchers(_default);
+        }
+    }
+
+    constexpr static auto noop_validator = [](const auto&) {
+        return std::nullopt;
+    };
+
 protected:
-    bool update_value(T&& new_value) {
-        if (new_value != _value) {
-            std::exception_ptr ex;
-            for (auto& binding : _bindings) {
-                try {
-                    binding.update(new_value);
-                } catch (...) {
-                    // In case there are multiple bindings:
-                    // if one of them throws an exception from an on_change
-                    // callback, proceed to update all bindings' values before
-                    // re-raising the last exception we saw.  This avoids
-                    // a situation where bindings could disagree about
-                    // the property's value.
-                    ex = std::current_exception();
-                }
+    void notify_watchers(const T& new_value) {
+        std::exception_ptr ex;
+        for (auto& binding : _bindings) {
+            try {
+                binding.update(new_value);
+            } catch (...) {
+                // In case there are multiple bindings:
+                // if one of them throws an exception from an on_change
+                // callback, proceed to update all bindings' values before
+                // re-raising the last exception we saw.  This avoids
+                // a situation where bindings could disagree about
+                // the property's value.
+                ex = std::current_exception();
             }
+        }
 
-            if (ex) {
-                rethrow_exception(ex);
-            }
+        if (ex) {
+            rethrow_exception(ex);
+        }
+    }
 
+    bool update_value(T&& new_value) {
+        if (new_value != _value) {
+            notify_watchers(new_value);
             _value = std::move(new_value);
 
             return true;
@@ -209,13 +254,14 @@ class property : public base_property {
     }
 
     T _value;
-    const T _default;
+    T _default;
+
+    // An alternative default that applies if the cluster's original logical
+    // version is <= the defined version
+    const std::optional<legacy_default<T>> _legacy_default;
 
 private:
     validator _validator;
-    constexpr static auto noop_validator = [](const auto&) {
-        return std::nullopt;
-    };
 
     friend class binding<T>;
     friend class mock_property<T>;

src/v/features/CMakeLists.txt

@@ -10,6 +10,7 @@ v_cc_library(
   DEPS
     Seastar::seastar
     v::model
+    v::config
   )
 
 add_dependencies(v_features kafka_codegen_headers)

src/v/features/feature_table.cc

@@ -242,7 +242,7 @@ void feature_table::set_active_version(
       && _original_version == invalid_version) {
         // Rely on controller log replay to call us first with
         // the first version the cluster ever agreed upon.
-        _original_version = v;
+        set_original_version(v);
     }
 
     for (auto& fs : _feature_state) {
@@ -296,7 +296,7 @@ void feature_table::bootstrap_original_version(cluster_version v) {
           v);
     }
 
-    _original_version = v;
+    set_original_version(v);
 
     // No on_update() call needed: bootstrap version is only advisory and
     // does not drive the feature state machines.
@@ -520,6 +520,14 @@ feature_table::decode_version_fence(model::record_batch batch) {
     return serde::from_iobuf<version_fence>(rec.release_value());
 }
 
+void feature_table::set_original_version(cluster::cluster_version v) {
+    _original_version = v;
+    if (v != cluster::invalid_version) {
+        config::shard_local_cfg().notify_original_version(
+          config::legacy_version{v});
+    }
+}
+
 } // namespace features
 
 namespace cluster {

src/v/features/feature_table.h

@@ -443,9 +443,7 @@ class feature_table {
 
     // The controller log offset of last batch applied to this state machine
     void set_applied_offset(model::offset o) { _applied_offset = o; }
-    void set_original_version(cluster::cluster_version v) {
-        _original_version = v;
-    }
+    void set_original_version(cluster::cluster_version v);
 
     void on_update();
 

tests/rptest/tests/resource_limits_test.py

@@ -88,16 +88,18 @@ def test_cpu_limited(self):
 
         rpk = RpkTool(self.redpanda)
 
-        # Three nodes, each with 1 core, 7000 partition-replicas
-        # per core, so with replicas=3, 7000 partitions should be the limit
+        # Three nodes, each with 1 core, 1000 partition-replicas
+        # per core, so with replicas=3, 1000 partitions should be the limit
         try:
-            rpk.create_topic("toobig", partitions=8000, replicas=3)
+            rpk.create_topic("toobig", partitions=1500, replicas=3)
         except RpkException as e:
             assert 'INVALID_PARTITIONS' in e.msg
         else:
             assert False
 
-        rpk.create_topic("okay", partitions=6000, replicas=3)
+        # This is not exactly 1000 because of system partitions consuming
+        # some of hte allowance.
+        rpk.create_topic("okay", partitions=900, replicas=3)
 
     @cluster(num_nodes=3)
     def test_fd_limited(self):