Merge pull request #24107 from oleiman/euse/core-8102/schema-validati…

…on-sanctions-alter-followup [CORE-8102] Separate key and value schema ID validation sanctions
redpanda-data · Nov 14, 2024 · 6924b5b · 6924b5b
2 parents 39638eb + 2856989
commit 6924b5b
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 5 deletions.
diff --git a/src/v/cluster/topics_frontend.cc b/src/v/cluster/topics_frontend.cc
@@ -116,14 +116,24 @@ std::vector<std::string_view> get_enterprise_features(
         features.emplace_back("tiered storage");
     }
 
-    constexpr auto schema_id_validation_enabled =
+    static constexpr auto key_schema_id_validation_enabled =
       [](const cluster::topic_properties& pp) -> bool {
         return pp.record_key_schema_id_validation.value_or(false)
-               || pp.record_key_schema_id_validation_compat.value_or(false)
-               || pp.record_value_schema_id_validation.value_or(false)
+               || pp.record_key_schema_id_validation_compat.value_or(false);
+    };
+
+    static constexpr auto value_schema_id_validation_enabled =
+      [](const cluster::topic_properties& pp) -> bool {
+        return pp.record_value_schema_id_validation.value_or(false)
                || pp.record_value_schema_id_validation_compat.value_or(false);
     };
 
+    static constexpr auto schema_id_validation_enabled =
+      [](const cluster::topic_properties& pp) -> bool {
+        return key_schema_id_validation_enabled(pp)
+               || value_schema_id_validation_enabled(pp);
+    };
+
     constexpr auto unset_or_unchanged =
       [](
         const reflection::is_std_optional auto& curr,
@@ -152,8 +162,9 @@ std::vector<std::string_view> get_enterprise_features(
     };
 
     if (
-      (schema_id_validation_enabled(properties)
-       < schema_id_validation_enabled(updated_properties))
+      ((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");
     }

diff --git a/src/v/kafka/server/tests/alter_config_test.cc b/src/v/kafka/server/tests/alter_config_test.cc
@@ -1396,6 +1396,40 @@ FIXTURE_TEST(test_unlicensed_alter_configs, alter_config_test_fixture) {
           },
           failure);
 
+        const auto key_validation = props_t{
+          with(kafka::topic_property_record_key_schema_id_validation, true),
+        };
+        test_cases.emplace_back(
+          "set_value_after_key",
+          key_validation,
+          alter_props_t{{set(
+            kafka::topic_property_record_value_schema_id_validation_compat,
+            true)}},
+          failure);
+        test_cases.emplace_back(
+          "unset_key",
+          key_validation,
+          alter_props_t{{set(
+            kafka::topic_property_record_key_schema_id_validation, false)}},
+          success);
+
+        const auto value_validation = props_t{
+          with(kafka::topic_property_record_value_schema_id_validation, true),
+        };
+        test_cases.emplace_back(
+          "set_key_after_value",
+          value_validation,
+          alter_props_t{{set(
+            kafka::topic_property_record_key_schema_id_validation_compat,
+            true)}},
+          failure);
+        test_cases.emplace_back(
+          "unset_value",
+          value_validation,
+          alter_props_t{{set(
+            kafka::topic_property_record_value_schema_id_validation, false)}},
+          success);
+
         const auto validation_with_strat = props_t{
           with(kafka::topic_property_record_key_schema_id_validation, true),
           with(kafka::topic_property_record_value_schema_id_validation, true),