[CORE-16608] cluster_link: validate source SR reachability and target emptiness on link creation

src/v/cluster/BUILD

@@ -2441,6 +2441,7 @@ redpanda_cc_library(
         "//src/v/cloud_topics/level_zero/stm:ctp_stm",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm_api",
         "//src/v/cluster_link/model",
+        "//src/v/cluster_link/model:sr_context_mapping",
         "//src/v/config",
         "//src/v/container:chunked_hash_map",
         "//src/v/container:chunked_vector",

src/v/cluster/cluster_link/frontend.cc

@@ -18,6 +18,7 @@
 #include "cluster/partition_leaders_table.h"
 #include "cluster/types.h"
 #include "cluster_link/model/filter_utils.h"
+#include "cluster_link/model/sr_context_mapping.h"
 #include "cluster_link/model/types.h"
 #include "config/configuration.h"
 #include "model/namespace.h"
@@ -339,32 +340,14 @@ bool link_shadows_schema_registry(const ::cluster_link::model::metadata& md) {
            || md.configuration.schema_registry_sync_cfg.api_mode() != nullptr;
 }
 
-bool filter_selects_source_context(
-  const ::cluster_link::model::schema_registry_sync_config::source_filter&
-    filter,
-  const ppsr::context& source_context) {
-    // An empty filter selects every source context.
-    if (filter.contexts.empty() && filter.subjects.empty()) {
-        return true;
-    }
-    if (std::ranges::contains(filter.contexts, source_context())) {
-        return true;
-    }
-    return std::ranges::any_of(
-      filter.subjects, [&source_context](const auto& subject) {
-          const auto parsed = ppsr::context_subject::from_string(
-            subject, ppsr::qualified_subjects_enabled::yes);
-          return parsed.ctx == source_context;
-      });
-}
-
 bool api_mode_shadows_context(
   const ::cluster_link::model::schema_registry_sync_config::
     shadow_schema_registry_api& api,
   const ppsr::context& dest_context) {
     // Identity mapping: the destination context name equals the source name.
     if (!api.destination) {
-        return filter_selects_source_context(api.filter, dest_context);
+        return ::cluster_link::model::filter_selects_source_context(
+          api.filter, dest_context);
     }
 
     return ss::visit(
@@ -374,7 +357,8 @@ bool api_mode_shadows_context(
           identity_context_mapping&) {
           // Identity mapping: the destination context name equals the source
           // name.
-          return filter_selects_source_context(api.filter, dest_context);
+          return ::cluster_link::model::filter_selects_source_context(
+            api.filter, dest_context);
       },
       [&api, &dest_context](
         const ::cluster_link::model::schema_registry_sync_config::
@@ -387,7 +371,7 @@ bool api_mode_shadows_context(
           for (const auto& [src_ctx, dst_ctx] : destination.mappings) {
               if (
                 dest_context == dst_ctx
-                && filter_selects_source_context(
+                && ::cluster_link::model::filter_selects_source_context(
                   api.filter, ppsr::context{src_ctx})) {
                   return true;
               }

src/v/cluster_link/BUILD

@@ -65,6 +65,7 @@ redpanda_cc_library(
         "link_probe.cc",
         "link_status_reconciler.cc",
         "manager.cc",
+        "sr_preflight_checker.cc",
         "task.cc",
         "topic_reconciler.cc",
     ],
@@ -74,6 +75,7 @@ redpanda_cc_library(
         "link_probe.h",
         "link_status_reconciler.h",
         "manager.h",
+        "sr_preflight_checker.h",
         "task.h",
         "topic_reconciler.h",
         "types.h",
@@ -83,8 +85,17 @@ redpanda_cc_library(
         "//src/v/cluster:feature_backend",
         "//src/v/cluster:plugin_backend",
         "//src/v/cluster:types",
+        "//src/v/cluster_link/model:sr_context_mapping",
+        "//src/v/cluster_link/schema_registry_sync:http_source_reader",
+        "//src/v/cluster_link/schema_registry_sync:source_reader",
         "//src/v/cluster_link/utils:topic_properties_utils",
+        "//src/v/container:chunked_hash_map",
         "//src/v/features",
+        "//src/v/pandaproxy/schema_registry/rest_client:client",
+        "//src/v/pandaproxy/schema_registry/rest_client:error",
+        "//src/v/schema:registry",
+        "//src/v/utils:retry_chain_node",
+        "@abseil-cpp//absl/container:flat_hash_set",
     ],
     visibility = ["//src/v/cluster_link:__subpackages__"],
     deps = [
@@ -111,6 +122,8 @@ redpanda_cc_library(
         "//src/v/kafka/data/rpc",
         "//src/v/metrics",
         "//src/v/model",
+        "//src/v/pandaproxy/schema_registry:types",
+        "//src/v/schema:fwd",
         "//src/v/security",
         "//src/v/ssx:future_util",
         "//src/v/ssx:mutex",

src/v/cluster_link/errc.cc

@@ -82,6 +82,12 @@ struct error_category final : public std::error_category {
             return "failed to stop task";
         case errc::failed_to_pause_task:
             return "failed to pause task";
+        case errc::link_sr_unreachable:
+            return "unable to reach source schema registry";
+        case errc::link_sr_target_not_empty:
+            return "target schema registry context is not empty";
+        case errc::link_sr_verification_failed:
+            return "schema registry verification failed";
         }
 
         return "(unknown error code)";

src/v/cluster_link/errc.h

@@ -51,6 +51,9 @@ enum class errc : int {
     link_verification_unknown_error,
     failed_to_stop_task,
     failed_to_pause_task,
+    link_sr_unreachable,
+    link_sr_target_not_empty,
+    link_sr_verification_failed,
 };
 
 std::error_code make_error_code(errc) noexcept;

src/v/cluster_link/fwd.h

@@ -15,4 +15,5 @@ namespace cluster_link {
 class link;
 class manager;
 class service;
+class sr_preflight_checker;
 } // namespace cluster_link

src/v/cluster_link/manager.cc

@@ -92,6 +92,7 @@ manager::manager(
   std::unique_ptr<partition_metadata_provider> partition_metadata_provider,
   std::unique_ptr<kafka_rpc_client_service> kafka_rpc_client_service,
   std::unique_ptr<members_table_provider> members_table_provider,
+  std::unique_ptr<sr_preflight_checker> sr_preflight,
   ss::lowres_clock::duration task_reconciler_interval,
   config::binding<int16_t> default_topic_replication,
   ss::scheduling_group scheduling_group)
@@ -108,6 +109,7 @@ manager::manager(
   , _partition_metadata_provider(std::move(partition_metadata_provider))
   , _kafka_rpc_client_service(std::move(kafka_rpc_client_service))
   , _members_table_provider(std::move(members_table_provider))
+  , _sr_preflight(std::move(sr_preflight))
   , _queue(
       scheduling_group,
       [](const std::exception_ptr& ex) {
@@ -351,14 +353,13 @@ ss::future<err_info> manager::link_preflight_checks(const model::metadata& md) {
             e.what())};
     }
     co_await stop_and_ignore();
-    co_return return_error;
+    if (return_error.code() != errc::success) {
+        co_return return_error;
+    }
+    co_return co_await _sr_preflight->check(md, _as);
 }
 
-ss::future<cl_result<void>>
-manager::test_connection(model::name_t name, model::connection_config config) {
-    model::metadata md;
-    md.name = std::move(name);
-    md.connection = std::move(config);
+ss::future<cl_result<void>> manager::test_connection(model::metadata md) {
     auto err = co_await link_preflight_checks(md);
     if (err.code() != errc::success) {
         co_return err;

src/v/cluster_link/manager.h

@@ -17,6 +17,7 @@
 #include "cluster_link/link_status_reconciler.h"
 #include "cluster_link/logger.h"
 #include "cluster_link/model/types.h"
+#include "cluster_link/sr_preflight_checker.h"
 #include "cluster_link/task.h"
 #include "cluster_link/topic_reconciler.h"
 #include "cluster_link/types.h"
@@ -56,6 +57,7 @@ class manager {
       std::unique_ptr<partition_metadata_provider> partition_metadata_provider,
       std::unique_ptr<kafka_rpc_client_service> kafka_rpc_client_service,
       std::unique_ptr<members_table_provider> members_table_provider,
+      std::unique_ptr<sr_preflight_checker> sr_preflight,
       ss::lowres_clock::duration task_reconciler_interval,
       config::binding<int16_t> default_topic_replication,
       ss::scheduling_group scheduling_group);
@@ -214,8 +216,7 @@ class manager {
 
     members_table_provider& get_members_table_provider() noexcept;
 
-    ss::future<cl_result<void>>
-    test_connection(model::name_t name, model::connection_config config);
+    ss::future<cl_result<void>> test_connection(model::metadata md);
 
 private:
     /// Called periodically to reconcile registered tasks on created links
@@ -244,6 +245,7 @@ class manager {
     std::unique_ptr<partition_metadata_provider> _partition_metadata_provider;
     std::unique_ptr<kafka_rpc_client_service> _kafka_rpc_client_service;
     std::unique_ptr<members_table_provider> _members_table_provider;
+    std::unique_ptr<sr_preflight_checker> _sr_preflight;
     ssx::work_queue _queue;
 
     chunked_vector<std::unique_ptr<task_factory>> _task_factories;

src/v/cluster_link/model/BUILD

@@ -1,5 +1,20 @@
 load("//bazel:build.bzl", "redpanda_cc_library")
 
+redpanda_cc_library(
+    name = "sr_context_mapping",
+    srcs = [
+        "sr_context_mapping.cc",
+    ],
+    hdrs = [
+        "sr_context_mapping.h",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":model",
+        "//src/v/pandaproxy/schema_registry:types",
+    ],
+)
+
 redpanda_cc_library(
     name = "model",
     srcs = [

src/v/cluster_link/model/sr_context_mapping.cc

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#include "cluster_link/model/sr_context_mapping.h"
+
+#include <algorithm>
+
+namespace cluster_link::model {
+
+bool filter_selects_source_context(
+  const schema_registry_sync_config::source_filter& filter,
+  const pandaproxy::schema_registry::context& source_context) {
+    namespace ppsr = pandaproxy::schema_registry;
+    // An empty filter selects every source context.
+    if (filter.contexts.empty() && filter.subjects.empty()) {
+        return true;
+    }
+    if (std::ranges::contains(filter.contexts, source_context())) {
+        return true;
+    }
+    return std::ranges::any_of(
+      filter.subjects, [&source_context](const auto& subject) {
+          const auto parsed = ppsr::context_subject::from_string(
+            subject, ppsr::qualified_subjects_enabled::yes);
+          return parsed.ctx == source_context;
+      });
+}
+
+} // namespace cluster_link::model

src/v/cluster_link/model/sr_context_mapping.h

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+
+#include "cluster_link/model/types.h"
+#include "pandaproxy/schema_registry/types.h"
+
+namespace cluster_link::model {
+
+/// Returns true if the API-sync source filter selects \p source_context.
+///
+/// An empty filter (no contexts and no subjects) selects every source context.
+/// This is the single source of truth for filter membership, shared by the
+/// runtime client-write blocker (cluster/cluster_link/frontend.cc) and the
+/// creation-time Schema Registry preflight check
+/// (cluster_link/sr_preflight_checker.cc) so the two cannot drift.
+bool filter_selects_source_context(
+  const schema_registry_sync_config::source_filter& filter,
+  const pandaproxy::schema_registry::context& source_context);
+
+} // namespace cluster_link::model

src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h

@@ -350,6 +350,14 @@ class delegating_registry : public schema::registry {
       ppsr::include_deleted inc) const override {
         return _inner->list_subject_versions(std::move(filter), inc);
     }
+    ss::future<bool>
+    has_subjects(ppsr::context ctx, ppsr::include_deleted inc) const override {
+        return _inner->has_subjects(std::move(ctx), inc);
+    }
+    ss::future<chunked_vector<ppsr::context_subject>>
+    get_subjects(ppsr::include_deleted inc) const override {
+        return _inner->get_subjects(inc);
+    }
     ss::future<ppsr::context_schema_id>
     create_schema(ppsr::subject_schema s) override {
         return _inner->create_schema(std::move(s));

src/v/cluster_link/service.cc

@@ -35,6 +35,7 @@
 #include "cluster_link/security_migrator.h"
 #include "cluster_link/shadow_linking_rpc_service.h"
 #include "cluster_link/source_topic_syncer.h"
+#include "cluster_link/sr_preflight_checker.h"
 #include "config/node_config.h"
 #include "kafka/client/direct_consumer/direct_consumer.h"
 #include "kafka/data/make_exact_offset_replicator.h"
@@ -1172,12 +1173,10 @@ service::delete_cluster_link(model::name_t name, bool force_delete_link) {
       });
 }
 
-ss::future<cl_result<void>>
-service::test_connection(model::name_t name, model::connection_config config) {
+ss::future<cl_result<void>> service::test_connection(model::metadata md) {
     auto h = _gate.hold();
-    return with_manager([name = std::move(name),
-                         config = std::move(config)](manager* mgr) mutable {
-        return mgr->test_connection(std::move(name), std::move(config));
+    return with_manager([md = std::move(md)](manager* mgr) mutable {
+        return mgr->test_connection(std::move(md));
     });
 }
 
@@ -1276,6 +1275,12 @@ ss::future<> service::maybe_start_manager() {
     if (_manager) {
         co_return;
     }
+    // The destination Schema Registry is owned by the service so it outlives
+    // the manager and its tasks. Construct it before the manager so it can be
+    // handed in for the Schema Registry preflight checks.
+    _schema_registry_dest = schema::registry::make_default(
+      _schema_registry_api);
+
     _manager = std::make_unique<manager>(
       _self,
       partition_leader_cache::make_default(_partition_leaders_table),
@@ -1299,6 +1304,8 @@ ss::future<> service::maybe_start_manager() {
         _hm_frontend),
       kafka_rpc_client_service::make_default(_kafka_data_rpc_client),
       members_table_provider::make_default(&_controller->get_members_table()),
+      sr_preflight_checker::make_default(
+        *_schema_registry_dest, source_sr_prober::make_default()),
       30s, // Temporary until we have a proper configuration for this
       config::shard_local_cfg().default_topic_replication.bind(),
       _scheduling_group);
@@ -1307,11 +1314,9 @@ ss::future<> service::maybe_start_manager() {
     co_await _manager->register_task_factory<security_migrator_factory>();
     co_await _manager->register_task_factory<roles_migrator_factory>();
 
-    // The destination Schema Registry and source reader factory are owned by
-    // the service so they outlive the tasks. Each link's mirroring task asks
-    // the factory for an HTTP-backed reader bound to its configured source.
-    _schema_registry_dest = schema::registry::make_default(
-      _schema_registry_api);
+    // The source reader factory is owned by the service so it outlives the
+    // tasks. Each link's mirroring task asks the factory for an HTTP-backed
+    // reader bound to its configured source.
     _source_reader_factory
       = std::make_unique<schema_registry_sync::http_source_reader_factory>();
     co_await _manager