diff --git a/src/v/cluster/topics_frontend.cc b/src/v/cluster/topics_frontend.cc index 8d25f2361f90..0f4a85733c50 100644 --- a/src/v/cluster/topics_frontend.cc +++ b/src/v/cluster/topics_frontend.cc @@ -116,14 +116,24 @@ std::vector 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 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 index 4423a643bb7d..61527c6eb641 100644 --- 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),