WIP k/create_topics: iceberg_partition_spec property validation

redpanda-data · Jan 20, 2025 · 7d0f9c5 · 7d0f9c5
1 parent 7c4f227
commit 7d0f9c5
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 9 deletions.
diff --git a/src/v/kafka/server/BUILD b/src/v/kafka/server/BUILD
@@ -176,6 +176,7 @@ redpanda_cc_library(
         "//src/v/features",
         "//src/v/hashing:jump_consistent",
         "//src/v/hashing:xx",
+        "//src/v/iceberg:unresolved_partition_spec",
         "//src/v/kafka/data:partition_proxy",
         "//src/v/kafka/protocol",
         "//src/v/kafka/protocol:add_offsets_to_txn",

diff --git a/src/v/kafka/server/handlers/topics/validators.h b/src/v/kafka/server/handlers/topics/validators.h
@@ -11,6 +11,7 @@
 
 #pragma once
 #include "config/configuration.h"
+#include "iceberg/unresolved_partition_spec.h"
 #include "kafka/protocol/schemata/create_topics_request.h"
 #include "kafka/protocol/schemata/create_topics_response.h"
 #include "kafka/server/handlers/topics/types.h"
@@ -276,22 +277,37 @@ struct iceberg_config_validator {
     static constexpr error_code ec = error_code::invalid_config;
 
     static bool is_valid(const creatable_topic& c) {
-        auto it = std::find_if(
+        model::iceberg_mode parsed_mode = model::iceberg_mode::disabled;
+
+        auto mode_it = std::find_if(
           c.configs.begin(),
           c.configs.end(),
           [](const createable_topic_config& cfg) {
               return cfg.name == topic_property_iceberg_mode;
           });
-        if (it == c.configs.end() || !it->value.has_value()) {
-            return true;
+        if (mode_it != c.configs.end() && mode_it->value.has_value()) {
+            try {
+                parsed_mode = boost::lexical_cast<model::iceberg_mode>(
+                  mode_it->value.value());
+            } catch (...) {
+                return false;
+            }
         }
-        model::iceberg_mode parsed_mode;
-        try {
-            parsed_mode = boost::lexical_cast<model::iceberg_mode>(
-              it->value.value());
-        } catch (...) {
-            return false;
+
+        auto pspec_it = std::find_if(
+          c.configs.begin(),
+          c.configs.end(),
+          [](const createable_topic_config& cfg) {
+              return cfg.name == topic_property_iceberg_partition_spec;
+          });
+        if (pspec_it != c.configs.end() && pspec_it->value.has_value()) {
+            auto parsed = iceberg::unresolved_partition_spec::parse(
+              pspec_it->value.value());
+            if (!parsed.has_value()) {
+                return false;
+            }
         }
+
         // If iceberg is enabled at the cluster level, the topic can
         // be created with any override. If it is disabled
         // at the cluster level, it cannot be enabled with a topic