cluster: remove some data policy bits

src/v/cluster/commands.h

@@ -112,10 +112,6 @@ static constexpr int8_t create_role_cmd_type = 10;
 static constexpr int8_t delete_role_cmd_type = 11;
 static constexpr int8_t update_role_cmd_type = 12;
 
-// data policy commands
-static constexpr int8_t create_data_policy_cmd_type = 0;
-static constexpr int8_t delete_data_policy_cmd_type = 1;
-
 // node management commands
 static constexpr int8_t decommission_node_cmd_type = 0;
 static constexpr int8_t recommission_node_cmd_type = 1;
@@ -309,20 +305,6 @@ using delete_acls_cmd = controller_command<
   model::record_batch_type::acl_management_cmd,
   serde_opts::adl_and_serde>;
 
-using create_data_policy_cmd = controller_command<
-  model::topic_namespace,
-  create_data_policy_cmd_data,
-  create_data_policy_cmd_type,
-  model::record_batch_type::data_policy_management_cmd,
-  serde_opts::adl_and_serde>;
-
-using delete_data_policy_cmd = controller_command<
-  model::topic_namespace,
-  std::optional<ss::sstring>,
-  delete_data_policy_cmd_type,
-  model::record_batch_type::data_policy_management_cmd,
-  serde_opts::adl_and_serde>;
-
 using decommission_node_cmd = controller_command<
   model::node_id,
   int8_t, // unused

src/v/cluster/controller_log_limiter.h

@@ -94,8 +94,6 @@ class controller_log_limiter {
           std::is_same_v<Cmd, update_role_cmd>) {
             return _acls_and_users_operations_limiter.try_throttle();
         } else if constexpr (
-          std::is_same_v<Cmd, create_data_policy_cmd> ||   //
-          std::is_same_v<Cmd, delete_data_policy_cmd> ||   //
           std::is_same_v<Cmd, cluster_config_delta_cmd> || //
           std::is_same_v<Cmd, feature_update_license_update_cmd>) {
             return _configuration_operations_limiter.try_throttle();

src/v/cluster/tests/serialization_rt_test.cc

@@ -902,14 +902,6 @@ SEASTAR_THREAD_TEST_CASE(serde_reflection_roundtrip) {
         }
         roundtrip_test(data);
     }
-    {
-        cluster::create_data_policy_cmd_data data;
-        data.dp = v8_engine::data_policy(
-          random_generators::gen_alphanum_string(20),
-          random_generators::gen_alphanum_string(20));
-
-        roundtrip_test(data);
-    }
     {
         cluster::config_status status;
         status.node = tests::random_named_int<model::node_id>();
@@ -2260,17 +2252,6 @@ SEASTAR_THREAD_TEST_CASE(commands_serialization_test) {
         }
 
         roundtrip_cmd<cluster::delete_acls_cmd>(std::move(delete_acl_data), 0);
-        cluster::create_data_policy_cmd_data create_dp;
-        create_dp.dp = v8_engine::data_policy(
-          random_generators::gen_alphanum_string(15),
-          random_generators::gen_alphanum_string(15));
-
-        roundtrip_cmd<cluster::create_data_policy_cmd>(
-          model::random_topic_namespace(), std::move(create_dp));
-
-        roundtrip_cmd<cluster::delete_data_policy_cmd>(
-          model::random_topic_namespace(),
-          random_generators::gen_alphanum_string(20));
 
         roundtrip_cmd<cluster::decommission_node_cmd>(
           tests::random_named_int<model::node_id>(), 0);

src/v/cluster/types.cc

@@ -1141,24 +1141,6 @@ adl<cluster::create_partitions_configuration>::from(iobuf_parser& in) {
     return ret;
 }
 
-void adl<cluster::create_data_policy_cmd_data>::to(
-  iobuf& out, cluster::create_data_policy_cmd_data&& dp_cmd_data) {
-    return serialize(
-      out, dp_cmd_data.current_version, std::move(dp_cmd_data.dp));
-}
-
-cluster::create_data_policy_cmd_data
-adl<cluster::create_data_policy_cmd_data>::from(iobuf_parser& in) {
-    auto version = adl<int8_t>{}.from(in);
-    vassert(
-      version == cluster::create_data_policy_cmd_data::current_version,
-      "Unexpected set_data_policy_cmd version {} (expected {})",
-      version,
-      cluster::create_data_policy_cmd_data::current_version);
-    auto dp = adl<v8_engine::data_policy>{}.from(in);
-    return cluster::create_data_policy_cmd_data{.dp = std::move(dp)};
-}
-
 void adl<cluster::incremental_topic_updates>::to(
   iobuf& out, cluster::incremental_topic_updates&& t) {
     reflection::serialize(

src/v/cluster/types.h

@@ -1461,22 +1461,6 @@ struct backend_operation
     }
 };
 
-struct create_data_policy_cmd_data
-  : serde::envelope<
-      create_data_policy_cmd_data,
-      serde::version<0>,
-      serde::compat_version<0>> {
-    static constexpr int8_t current_version = 1; // In future dp will be vector
-
-    auto serde_fields() { return std::tie(dp); }
-
-    v8_engine::data_policy dp;
-
-    friend bool operator==(
-      const create_data_policy_cmd_data&, const create_data_policy_cmd_data&)
-      = default;
-};
-
 using config_version = named_type<int64_t, struct config_version_type>;
 constexpr config_version config_version_unset = config_version{-1};
 
@@ -3508,12 +3492,6 @@ struct adl<cluster::create_partitions_configuration> {
     cluster::create_partitions_configuration from(iobuf_parser&);
 };
 
-template<>
-struct adl<cluster::create_data_policy_cmd_data> {
-    void to(iobuf&, cluster::create_data_policy_cmd_data&&);
-    cluster::create_data_policy_cmd_data from(iobuf_parser&);
-};
-
 template<>
 struct adl<cluster::incremental_topic_updates> {
     void to(iobuf& out, cluster::incremental_topic_updates&&);

src/v/kafka/server/handlers/alter_configs.cc

@@ -303,16 +303,6 @@ create_topic_properties_update(
               error_code::invalid_config,
               fmt::format(
                 "unable to parse property {} value {}", cfg.name, cfg.value));
-        } catch (const v8_engine::data_policy_exeption& e) {
-            return make_error_alter_config_resource_response<
-              alter_configs_resource_response>(
-              resource,
-              error_code::invalid_config,
-              fmt::format(
-                "unable to parse property {}, value{}, error {}",
-                cfg.name,
-                cfg.value,
-                e.what()));
         }
 
         // Unsupported property, return error

src/v/kafka/server/handlers/configs/config_response_utils.cc

@@ -100,7 +100,6 @@ consteval describe_configs_type property_config_type() {
         std::is_same_v<T, model::cleanup_policy_bitflags> ||
         std::is_same_v<T, model::timestamp_type> ||
         std::is_same_v<T, config::data_directory_path> ||
-        std::is_same_v<T, v8_engine::data_policy> ||
         std::is_same_v<T, pandaproxy::schema_registry::subject_name_strategy> || 
         std::is_same_v<T, model::vcluster_id> ||
         std::is_same_v<T, model::write_caching_mode>;

src/v/v8_engine/data_policy.h

@@ -18,17 +18,6 @@
 #include <seastar/core/sstring.hh>
 namespace v8_engine {
 
-class data_policy_exeption final : public std::exception {
-public:
-    explicit data_policy_exeption(ss::sstring msg) noexcept
-      : _msg(std::move(msg)) {}
-
-    const char* what() const noexcept final { return _msg.c_str(); }
-
-private:
-    ss::sstring _msg;
-};
-
 // Datapolicy property for v8_engine. In first version it contains only
 // function_name and script_name, in the future it will contain ACLs, geo,
 // e.t.c.