[v26.1.x] datalake: validate iceberg_rest_catalog_endpoint at config time

src/v/config/BUILD

@@ -59,6 +59,7 @@ redpanda_cc_library(
     implementation_deps = [
         "//src/v/cluster:topic_memory_per_partition_default",
         "//src/v/datalake:partition_spec_parser",
+        "//src/v/datalake:validators",
         "@openssl",
     ],
     visibility = ["//visibility:public"],

src/v/config/configuration.cc

@@ -4147,7 +4147,8 @@ configuration::configuration()
         .example = "http://hostname:8181",
         .visibility = visibility::user,
       },
-      std::nullopt)
+      std::nullopt,
+      &validate_iceberg_rest_catalog_endpoint)
   , iceberg_rest_catalog_client_id(
       *this,
       "iceberg_rest_catalog_client_id",

src/v/config/validators.cc

@@ -17,6 +17,7 @@
 #include "config/sasl_mechanisms.h"
 #include "config/types.h"
 #include "datalake/partition_spec_parser.h"
+#include "datalake/validators.h"
 #include "model/namespace.h"
 #include "model/validation.h"
 #include "security/oidc_url_parser.h"
@@ -269,6 +270,19 @@ validate_iceberg_partition_spec(const ss::sstring& value) {
     return std::nullopt;
 }
 
+std::optional<ss::sstring> validate_iceberg_rest_catalog_endpoint(
+  const std::optional<ss::sstring>& endpoint) {
+    if (!endpoint.has_value()) {
+        return std::nullopt;
+    }
+    auto parsed = datalake::parse_iceberg_rest_catalog_endpoint(
+      endpoint.value());
+    if (!parsed.has_value()) {
+        return std::move(parsed).error();
+    }
+    return std::nullopt;
+}
+
 std::optional<ss::sstring> validate_iceberg_topic_name_dot_replacement(
   const std::optional<ss::sstring>& value) {
     if (value.has_value() && value->find('.') != ss::sstring::npos) {

src/v/config/validators.h

@@ -63,6 +63,9 @@ std::optional<ss::sstring> validate_tombstone_retention_ms(
 std::optional<ss::sstring>
 validate_iceberg_partition_spec(const ss::sstring& spec);
 
+std::optional<ss::sstring> validate_iceberg_rest_catalog_endpoint(
+  const std::optional<ss::sstring>& endpoint);
+
 std::optional<ss::sstring> validate_iceberg_topic_name_dot_replacement(
   const std::optional<ss::sstring>& value);
 

src/v/datalake/BUILD

@@ -204,6 +204,26 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "validators",
+    srcs = [
+        "validators.cc",
+    ],
+    hdrs = [
+        "validators.h",
+    ],
+    implementation_deps = [
+        "@abseil-cpp//absl/strings",
+        "@fmt",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        "//src/v/base",
+        "@ada",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "translation_task",
     srcs = [

src/v/datalake/coordinator/BUILD

@@ -124,6 +124,7 @@ redpanda_cc_library(
         "//src/v/base",
         "//src/v/config",
         "//src/v/datalake:logger",
+        "//src/v/datalake:validators",
         "//src/v/iceberg:catalog",
         "//src/v/iceberg:filesystem_catalog",
         "//src/v/iceberg:rest_catalog",

src/v/datalake/coordinator/catalog_factory.cc

@@ -14,6 +14,7 @@
 #include "config/types.h"
 #include "datalake/credential_manager.h"
 #include "datalake/logger.h"
+#include "datalake/validators.h"
 #include "iceberg/catalog.h"
 #include "iceberg/filesystem_catalog.h"
 #include "iceberg/rest_catalog.h"
@@ -78,38 +79,28 @@ struct endpoint_information {
 };
 
 endpoint_information endpoint_to_address(const ss::sstring& url_str) {
-    auto url = ada::parse(url_str);
-    if (!url) {
-        throw std::invalid_argument(
-          fmt::format(
-            "Malformed Iceberg REST catalog endpoint url: {}", url_str));
+    auto parsed = datalake::parse_iceberg_rest_catalog_endpoint(url_str);
+    if (!parsed.has_value()) {
+        throw std::invalid_argument(std::string(parsed.error()));
     }
+    const auto& url = parsed.value();
     // Default port as used by the Iceberg catalogs
-    uint16_t port = url->type == ada::scheme::HTTPS ? 443 : 8181;
-    if (url->has_port()) {
+    uint16_t port = url.type == ada::scheme::HTTPS ? 443 : 8181;
+    if (url.has_port()) {
         int32_t port_from_uri{0};
-        auto parsed = absl::SimpleAtoi(url->get_port(), &port_from_uri);
-        if (
-          !parsed || port_from_uri < 0
-          || port_from_uri > std::numeric_limits<uint16_t>::max()) {
-            throw std::invalid_argument(
-              fmt::format(
-                "Malformed Iceberg REST catalog endpoint url: {}, unable to "
-                "parse port",
-                url_str));
-        }
+        // Already validated above; parse cannot fail.
+        std::ignore = absl::SimpleAtoi(url.get_port(), &port_from_uri);
         port = static_cast<uint16_t>(port_from_uri);
     }
     std::optional<iceberg::rest_client::base_path> path
-      = url->get_pathname() != ""
+      = url.get_pathname() != ""
           ? std::make_optional<iceberg::rest_client::base_path>(
-              url->get_pathname())
+              url.get_pathname())
           : std::nullopt;
 
     return {
-      .address
-      = net::unresolved_address{ss::sstring(url->get_hostname()), port},
-      .needs_tls = url->type == ada::scheme::HTTPS,
+      .address = net::unresolved_address{ss::sstring(url.get_hostname()), port},
+      .needs_tls = url.type == ada::scheme::HTTPS,
       .base_path = std::move(path)};
 }
 

src/v/datalake/validators.cc

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+#include "datalake/validators.h"
+
+#include "absl/strings/numbers.h"
+
+#include <fmt/format.h>
+
+#include <ada.h>
+#include <limits>
+
+namespace datalake {
+
+tl::expected<ada::url_aggregator, ss::sstring>
+parse_iceberg_rest_catalog_endpoint(const ss::sstring& url_str) {
+    auto url = ada::parse(url_str);
+    if (!url.has_value()) {
+        return tl::unexpected(
+          fmt::format(
+            "Malformed Iceberg REST catalog endpoint url: {}", url_str));
+    }
+    if (url->has_port()) {
+        int32_t port_from_uri{0};
+        auto parsed = absl::SimpleAtoi(url->get_port(), &port_from_uri);
+        if (
+          !parsed || port_from_uri < 0
+          || port_from_uri > std::numeric_limits<uint16_t>::max()) {
+            return tl::unexpected(
+              fmt::format(
+                "Malformed Iceberg REST catalog endpoint url: {}, unable to "
+                "parse port",
+                url_str));
+        }
+    }
+    return std::move(url.value());
+}
+
+} // namespace datalake

src/v/datalake/validators.h

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+#pragma once
+
+#include "base/seastarx.h"
+
+#include <seastar/core/sstring.hh>
+
+#include <ada.h>
+
+namespace datalake {
+
+/// Parses and validates an `iceberg_rest_catalog_endpoint` value.
+/// Returns the parsed URL on success, or an error message describing why
+/// parsing failed.
+tl::expected<ada::url_aggregator, ss::sstring>
+parse_iceberg_rest_catalog_endpoint(const ss::sstring& url_str);
+
+} // namespace datalake

tests/rptest/tests/cluster_config_test.py

@@ -2069,6 +2069,40 @@ def test_iceberg_rest_catalog_endpoint_validation(self):
             expect_restart=True,
         )
 
+    @cluster(num_nodes=1)
+    def test_iceberg_rest_catalog_endpoint_url_validation(self):
+        """
+        Verifies that malformed `iceberg_rest_catalog_endpoint` values are
+        rejected, while well-formed URLs are accepted.
+        """
+        malformed_values = [
+            "not a url",
+            "://missing-scheme",
+            "http://host:not-a-port",
+            "http://host:99999",
+        ]
+        for value in malformed_values:
+            with expect_exception(
+                requests.exceptions.HTTPError,
+                lambda e: e.response.status_code == 400,
+            ):
+                self.redpanda.set_cluster_config(
+                    {"iceberg_rest_catalog_endpoint": value},
+                    expect_restart=True,
+                )
+
+        # Well-formed values should be accepted.
+        valid_values = [
+            "http://localhost:8181",
+            "https://catalog.example.com",
+            "https://catalog.example.com:443/path",
+        ]
+        for value in valid_values:
+            self.redpanda.set_cluster_config(
+                {"iceberg_rest_catalog_endpoint": value},
+                expect_restart=True,
+            )
+
 
 class PropertyAliasData(NamedTuple):
     """Data structure for property alias testing configuration."""