Merge pull request #24570 from michael-redpanda/bugs/core-8533/relax-…

…topic-configs [CORE-8533] relax topic configs
redpanda-data · Dec 16, 2024 · 3970cab · 3970cab
2 parents 56a2c8a + 98a08b2
commit 3970cab
Show file tree

Hide file tree

Showing 7 changed files with 636 additions and 357 deletions.
diff --git a/src/v/cluster/topics_frontend.cc b/src/v/cluster/topics_frontend.cc
@@ -74,18 +74,30 @@ std::vector<std::string_view>
 get_enterprise_features(const cluster::topic_configuration& cfg) {
     std::vector<std::string_view> features;
     const auto si_disabled = model::shadow_indexing_mode::disabled;
-    if (cfg.properties.shadow_indexing.value_or(si_disabled) != si_disabled) {
-        features.emplace_back("tiered storage");
-    }
-    if (cfg.is_recovery_enabled()) {
-        features.emplace_back("topic recovery");
-    }
-    if (cfg.is_read_replica()) {
-        features.emplace_back("remote read replicas");
+    // Only enforce tiered storage topic config sanctions when cloud storage is
+    // enabled for the cluster
+    if (config::shard_local_cfg().cloud_storage_enabled.is_restricted()) {
+        if (
+          cfg.properties.shadow_indexing.value_or(si_disabled) != si_disabled) {
+            features.emplace_back("tiered storage");
+        }
+        if (cfg.is_recovery_enabled()) {
+            features.emplace_back("topic recovery");
+        }
+        if (cfg.is_read_replica()) {
+            features.emplace_back("remote read replicas");
+        }
     }
-    if (cfg.is_schema_id_validation_enabled()) {
-        features.emplace_back("schema ID validation");
+
+    // Only enforce schema ID validation topic configs if Schema ID validation
+    // is enabled for the cluster
+    if (config::shard_local_cfg().enable_schema_id_validation.is_restricted()) {
+        if (cfg.is_schema_id_validation_enabled()) {
+            features.emplace_back("schema ID validation");
+        }
     }
+
+    // We are always enforcing leadership preference restrictions
     if (const auto& leaders_pref = cfg.properties.leaders_preference;
         leaders_pref.has_value()
         && config::shard_local_cfg()
@@ -111,11 +123,15 @@ std::vector<std::string_view> get_enterprise_features(
 
     std::vector<std::string_view> features;
     const auto si_disabled = model::shadow_indexing_mode::disabled;
-    if (
-      (properties.shadow_indexing.value_or(si_disabled)
-       < updated_properties.shadow_indexing.value_or(si_disabled))
-      || (properties.remote_delete < updated_properties.remote_delete)) {
-        features.emplace_back("tiered storage");
+    // Only enforce tiered storage topic config sanctions when cloud storage is
+    // enabled for the cluster
+    if (config::shard_local_cfg().cloud_storage_enabled.is_restricted()) {
+        if (
+          (properties.shadow_indexing.value_or(si_disabled)
+           < updated_properties.shadow_indexing.value_or(si_disabled))
+          || (properties.remote_delete < updated_properties.remote_delete)) {
+            features.emplace_back("tiered storage");
+        }
     }
 
     static constexpr auto key_schema_id_validation_enabled =
@@ -163,12 +179,16 @@ std::vector<std::string_view> get_enterprise_features(
             up.record_value_subject_name_strategy_compat));
     };
 
-    if (
-      ((key_schema_id_validation_enabled(properties)
-        < key_schema_id_validation_enabled(updated_properties))
-       || (value_schema_id_validation_enabled(properties) < value_schema_id_validation_enabled(updated_properties)))
-      || (schema_id_validation_enabled(updated_properties) && sns_modified())) {
-        features.emplace_back("schema id validation");
+    // Only enforce schema ID validation topic configs if Schema ID validation
+    // is enabled for the cluster
+    if (config::shard_local_cfg().enable_schema_id_validation.is_restricted()) {
+        if (
+          ((key_schema_id_validation_enabled(properties)
+            < key_schema_id_validation_enabled(updated_properties))
+           || (value_schema_id_validation_enabled(properties) < value_schema_id_validation_enabled(updated_properties)))
+          || (schema_id_validation_enabled(updated_properties) && sns_modified())) {
+            features.emplace_back("schema id validation");
+        }
     }
 
     if (const auto& updated_pref = updated_properties.leaders_preference;

diff --git a/src/v/config/property.h b/src/v/config/property.h
@@ -1091,6 +1091,11 @@ class enterprise : public P {
         return std::nullopt;
     }
 
+    /**
+     * @brief Checks current value of property to see if it is restricted
+     */
+    bool is_restricted() const { return do_check_restricted(this->value()); }
+
 private:
     bool do_check_restricted(const T& setting) const final {
         // depending on how the restriction was defined, construct an applicable

diff --git a/src/v/config/tests/enterprise_property_test.cc b/src/v/config/tests/enterprise_property_test.cc
@@ -109,4 +109,34 @@ TEST(EnterprisePropertyTest, TestTypeName) {
     EXPECT_EQ(cfg.enterprise_enum.type_name(), "string");
 }
 
+TEST(EnterprisePropertyTest, TestIsRestricted) {
+    test_config cfg;
+    cfg.enterprise_bool.set_value(false);
+    EXPECT_FALSE(cfg.enterprise_bool.is_restricted());
+    cfg.enterprise_bool.set_value(true);
+    EXPECT_TRUE(cfg.enterprise_bool.is_restricted());
+
+    cfg.enterprise_str_enum.set_value("foo");
+    EXPECT_FALSE(cfg.enterprise_str_enum.is_restricted());
+    cfg.enterprise_str_enum.set_value("bar");
+    EXPECT_TRUE(cfg.enterprise_str_enum.is_restricted());
+
+    cfg.enterprise_str_vec.set_value(
+      std::vector<ss::sstring>{"foo", "bar", "baz"});
+    EXPECT_FALSE(cfg.enterprise_str_vec.is_restricted());
+    cfg.enterprise_str_vec.set_value(
+      std::vector<ss::sstring>{"foo", "bar", "baz", "GSSAPI"});
+    EXPECT_TRUE(cfg.enterprise_str_vec.is_restricted());
+
+    cfg.enterprise_opt_int.set_value(10);
+    EXPECT_FALSE(cfg.enterprise_opt_int.is_restricted());
+    cfg.enterprise_opt_int.set_value(10000);
+    EXPECT_TRUE(cfg.enterprise_opt_int.is_restricted());
+
+    cfg.enterprise_enum.set_value(tls_version::v1_0);
+    EXPECT_FALSE(cfg.enterprise_enum.is_restricted());
+    cfg.enterprise_enum.set_value(tls_version::v1_3);
+    EXPECT_TRUE(cfg.enterprise_enum.is_restricted());
+}
+
 } // namespace config