From 3ed7d0dfc5cbada7e3f74cba59e832111dae43bd Mon Sep 17 00:00:00 2001 From: Mateusz Najda Date: Fri, 3 Jul 2026 09:06:04 +0200 Subject: [PATCH 1/5] schema: add has_subjects to the registry interface Adds a short-circuiting emptiness probe forwarding to the existing sharded_store::has_subjects, plus implementations in every registry subclass so the interface stays pure-virtual. --- .../schema_registry_sync/tests/sr_sync_test_fixtures.h | 4 ++++ src/v/datalake/tests/record_schema_resolver_test.cc | 6 ++++++ src/v/schema/registry.cc | 10 ++++++++++ src/v/schema/registry.h | 7 +++++++ src/v/schema/tests/fake_registry.cc | 8 ++++++++ src/v/schema/tests/fake_registry.h | 4 ++++ 6 files changed, 39 insertions(+) diff --git a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h index f566280d07b7c..59632a1225696 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h +++ b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h @@ -350,6 +350,10 @@ class delegating_registry : public schema::registry { ppsr::include_deleted inc) const override { return _inner->list_subject_versions(std::move(filter), inc); } + ss::future + has_subjects(ppsr::context ctx, ppsr::include_deleted inc) const override { + return _inner->has_subjects(std::move(ctx), inc); + } ss::future create_schema(ppsr::subject_schema s) override { return _inner->create_schema(std::move(s)); diff --git a/src/v/datalake/tests/record_schema_resolver_test.cc b/src/v/datalake/tests/record_schema_resolver_test.cc index 2665dfd4d8cf0..f97149d75ef9b 100644 --- a/src/v/datalake/tests/record_schema_resolver_test.cc +++ b/src/v/datalake/tests/record_schema_resolver_test.cc @@ -619,6 +619,12 @@ class counting_registry : public schema::registry { return _registry.list_subject_versions(std::move(filter), inc_del); } + ss::future has_subjects( + pandaproxy::schema_registry::context ctx, + pandaproxy::schema_registry::include_deleted inc_del) const override { + return _registry.has_subjects(std::move(ctx), inc_del); + } + ss::future create_schema( pandaproxy::schema_registry::subject_schema unparsed) override { return _registry.create_schema(std::move(unparsed)); diff --git a/src/v/schema/registry.cc b/src/v/schema/registry.cc index c896a38c1fb20..5c28e0810a76f 100644 --- a/src/v/schema/registry.cc +++ b/src/v/schema/registry.cc @@ -89,6 +89,11 @@ class schema_registry_impl : public registry { co_return co_await reader->list_subject_versions( std::move(filter), inc_del); } + ss::future has_subjects( + ppsr::context ctx, ppsr::include_deleted inc_del) const override { + auto [reader, _] = co_await service(); + co_return co_await reader->has_subjects(std::move(ctx), inc_del); + } ss::future create_schema(ppsr::subject_schema schema) override { @@ -224,6 +229,11 @@ class disabled_schema_registry : public registry { throw std::logic_error( "invalid attempted usage of a disabled schema registry"); } + ss::future + has_subjects(ppsr::context, ppsr::include_deleted) const override { + throw std::logic_error( + "invalid attempted usage of a disabled schema registry"); + } ss::future create_schema(ppsr::subject_schema) override { throw std::logic_error( diff --git a/src/v/schema/registry.h b/src/v/schema/registry.h index 01021130f912d..bc59b3bd88d2c 100644 --- a/src/v/schema/registry.h +++ b/src/v/schema/registry.h @@ -89,6 +89,13 @@ class registry { bool(const pandaproxy::schema_registry::context_subject&)> filter, pandaproxy::schema_registry::include_deleted) const = 0; + /// Returns true if \p ctx contains any subject. With include_deleted::yes, + /// soft-deleted subjects count. Short-circuits at the first match rather + /// than materializing the context's versions. + virtual ss::future has_subjects( + pandaproxy::schema_registry::context, + pandaproxy::schema_registry::include_deleted) const = 0; + virtual ss::future create_schema(pandaproxy::schema_registry::subject_schema) = 0; diff --git a/src/v/schema/tests/fake_registry.cc b/src/v/schema/tests/fake_registry.cc index b7604d8423e07..d46c51e22f1de 100644 --- a/src/v/schema/tests/fake_registry.cc +++ b/src/v/schema/tests/fake_registry.cc @@ -119,6 +119,14 @@ schema::fake_registry::list_subject_versions( co_return out; } +ss::future schema::fake_registry::has_subjects( + ppsr::context ctx, ppsr::include_deleted inc_del) const { + maybe_throw_injected_failure(); + co_return std::ranges::any_of(_store.schemas, [&](const auto& s) { + return s.schema.sub().ctx == ctx && (inc_del || !s.deleted); + }); +} + ss::future schema::fake_registry::create_schema(ppsr::subject_schema unparsed) { maybe_throw_injected_failure(); diff --git a/src/v/schema/tests/fake_registry.h b/src/v/schema/tests/fake_registry.h index d77d68fe8f3e7..b83c4c207e673 100644 --- a/src/v/schema/tests/fake_registry.h +++ b/src/v/schema/tests/fake_registry.h @@ -73,6 +73,10 @@ class fake_registry : public schema::registry { bool(const pandaproxy::schema_registry::context_subject&)> filter, pandaproxy::schema_registry::include_deleted inc_del) const override; + ss::future has_subjects( + pandaproxy::schema_registry::context ctx, + pandaproxy::schema_registry::include_deleted inc_del) const override; + ss::future create_schema( pandaproxy::schema_registry::subject_schema unparsed) override; From 4766575565c017fec4db70f9b5cc4a16bb10f224 Mon Sep 17 00:00:00 2001 From: Mateusz Najda Date: Tue, 7 Jul 2026 12:11:03 +0200 Subject: [PATCH 2/5] schema: add get_subjects to the registry interface Adds a subject listing forwarding to the existing sharded_store::get_subjects, returning one entry per subject rather than per (subject, version), plus implementations in every registry subclass so the interface stays pure-virtual. --- .../tests/sr_sync_test_fixtures.h | 4 ++++ src/v/datalake/tests/record_schema_resolver_test.cc | 6 ++++++ src/v/schema/registry.cc | 10 ++++++++++ src/v/schema/registry.h | 7 +++++++ src/v/schema/tests/BUILD | 1 + src/v/schema/tests/fake_registry.cc | 12 ++++++++++++ src/v/schema/tests/fake_registry.h | 4 ++++ 7 files changed, 44 insertions(+) diff --git a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h index 59632a1225696..06e237f07f427 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h +++ b/src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h @@ -354,6 +354,10 @@ class delegating_registry : public schema::registry { has_subjects(ppsr::context ctx, ppsr::include_deleted inc) const override { return _inner->has_subjects(std::move(ctx), inc); } + ss::future> + get_subjects(ppsr::include_deleted inc) const override { + return _inner->get_subjects(inc); + } ss::future create_schema(ppsr::subject_schema s) override { return _inner->create_schema(std::move(s)); diff --git a/src/v/datalake/tests/record_schema_resolver_test.cc b/src/v/datalake/tests/record_schema_resolver_test.cc index f97149d75ef9b..f4c9001c4623f 100644 --- a/src/v/datalake/tests/record_schema_resolver_test.cc +++ b/src/v/datalake/tests/record_schema_resolver_test.cc @@ -625,6 +625,12 @@ class counting_registry : public schema::registry { return _registry.has_subjects(std::move(ctx), inc_del); } + ss::future> + get_subjects( + pandaproxy::schema_registry::include_deleted inc_del) const override { + return _registry.get_subjects(inc_del); + } + ss::future create_schema( pandaproxy::schema_registry::subject_schema unparsed) override { return _registry.create_schema(std::move(unparsed)); diff --git a/src/v/schema/registry.cc b/src/v/schema/registry.cc index 5c28e0810a76f..acaa7bcb9cc5e 100644 --- a/src/v/schema/registry.cc +++ b/src/v/schema/registry.cc @@ -94,6 +94,11 @@ class schema_registry_impl : public registry { auto [reader, _] = co_await service(); co_return co_await reader->has_subjects(std::move(ctx), inc_del); } + ss::future> + get_subjects(ppsr::include_deleted inc_del) const override { + auto [reader, _] = co_await service(); + co_return co_await reader->get_subjects(inc_del, std::nullopt); + } ss::future create_schema(ppsr::subject_schema schema) override { @@ -234,6 +239,11 @@ class disabled_schema_registry : public registry { throw std::logic_error( "invalid attempted usage of a disabled schema registry"); } + ss::future> + get_subjects(ppsr::include_deleted) const override { + throw std::logic_error( + "invalid attempted usage of a disabled schema registry"); + } ss::future create_schema(ppsr::subject_schema) override { throw std::logic_error( diff --git a/src/v/schema/registry.h b/src/v/schema/registry.h index bc59b3bd88d2c..c669057fa7ede 100644 --- a/src/v/schema/registry.h +++ b/src/v/schema/registry.h @@ -96,6 +96,13 @@ class registry { pandaproxy::schema_registry::context, pandaproxy::schema_registry::include_deleted) const = 0; + /// Lists all subjects across all contexts, one entry per subject (no + /// version fan-out). With include_deleted::yes, soft-deleted subjects are + /// included. + virtual ss::future< + chunked_vector> + get_subjects(pandaproxy::schema_registry::include_deleted) const = 0; + virtual ss::future create_schema(pandaproxy::schema_registry::subject_schema) = 0; diff --git a/src/v/schema/tests/BUILD b/src/v/schema/tests/BUILD index be64aac9f2082..4a966db55373a 100644 --- a/src/v/schema/tests/BUILD +++ b/src/v/schema/tests/BUILD @@ -11,6 +11,7 @@ redpanda_test_cc_library( visibility = ["//visibility:public"], deps = [ "//src/v/base", + "//src/v/container:chunked_hash_map", "//src/v/pandaproxy/schema_registry:types", "//src/v/schema:registry", "@fmt", diff --git a/src/v/schema/tests/fake_registry.cc b/src/v/schema/tests/fake_registry.cc index d46c51e22f1de..76c96fe9fc075 100644 --- a/src/v/schema/tests/fake_registry.cc +++ b/src/v/schema/tests/fake_registry.cc @@ -11,6 +11,7 @@ #include "schema/tests/fake_registry.h" +#include "container/chunked_hash_map.h" #include "pandaproxy/schema_registry/errors.h" #include "pandaproxy/schema_registry/types.h" @@ -21,6 +22,7 @@ #include #include +#include #include namespace { @@ -127,6 +129,16 @@ ss::future schema::fake_registry::has_subjects( }); } +ss::future> +schema::fake_registry::get_subjects(ppsr::include_deleted inc_del) const { + maybe_throw_injected_failure(); + co_return _store.schemas | std::views::filter([inc_del](const auto& s) { + return inc_del || !s.deleted; + }) | std::views::transform([](const auto& s) { return s.schema.sub(); }) + | std::ranges::to>() + | std::ranges::to>(); +} + ss::future schema::fake_registry::create_schema(ppsr::subject_schema unparsed) { maybe_throw_injected_failure(); diff --git a/src/v/schema/tests/fake_registry.h b/src/v/schema/tests/fake_registry.h index b83c4c207e673..f0c2a4ccf8ed9 100644 --- a/src/v/schema/tests/fake_registry.h +++ b/src/v/schema/tests/fake_registry.h @@ -77,6 +77,10 @@ class fake_registry : public schema::registry { pandaproxy::schema_registry::context ctx, pandaproxy::schema_registry::include_deleted inc_del) const override; + ss::future> + get_subjects( + pandaproxy::schema_registry::include_deleted inc_del) const override; + ss::future create_schema( pandaproxy::schema_registry::subject_schema unparsed) override; From 3883449b751830fdcd4aa84698db846220e82973 Mon Sep 17 00:00:00 2001 From: Mateusz Najda Date: Fri, 3 Jul 2026 09:09:13 +0200 Subject: [PATCH 3/5] cluster_link: extract shared SR context-mapping predicate Lifts filter_selects_source_context out of frontend.cc into a shared model target so the runtime client-write blocker and the preflight check resolve filter membership the same way. Behavior-preserving. --- src/v/cluster/BUILD | 1 + src/v/cluster/cluster_link/frontend.cc | 28 +++----------- src/v/cluster_link/model/BUILD | 15 ++++++++ .../cluster_link/model/sr_context_mapping.cc | 37 +++++++++++++++++++ src/v/cluster_link/model/sr_context_mapping.h | 30 +++++++++++++++ 5 files changed, 89 insertions(+), 22 deletions(-) create mode 100644 src/v/cluster_link/model/sr_context_mapping.cc create mode 100644 src/v/cluster_link/model/sr_context_mapping.h diff --git a/src/v/cluster/BUILD b/src/v/cluster/BUILD index 000fc8f88a9e0..146d466e52d6a 100644 --- a/src/v/cluster/BUILD +++ b/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", diff --git a/src/v/cluster/cluster_link/frontend.cc b/src/v/cluster/cluster_link/frontend.cc index c60971f8c5149..0f528e3b93ba4 100644 --- a/src/v/cluster/cluster_link/frontend.cc +++ b/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; } diff --git a/src/v/cluster_link/model/BUILD b/src/v/cluster_link/model/BUILD index 79559d52e7693..541d7e64cd148 100644 --- a/src/v/cluster_link/model/BUILD +++ b/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 = [ diff --git a/src/v/cluster_link/model/sr_context_mapping.cc b/src/v/cluster_link/model/sr_context_mapping.cc new file mode 100644 index 0000000000000..c8522b1c10679 --- /dev/null +++ b/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 + +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 diff --git a/src/v/cluster_link/model/sr_context_mapping.h b/src/v/cluster_link/model/sr_context_mapping.h new file mode 100644 index 0000000000000..43b7196fecfe0 --- /dev/null +++ b/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 From be8b85c212eccc9b0f8e281966d85d418a2fafec Mon Sep 17 00:00:00 2001 From: Mateusz Najda Date: Fri, 3 Jul 2026 09:09:57 +0200 Subject: [PATCH 4/5] cluster_link: validate SR reachability and target emptiness on link creation Extends manager::link_preflight_checks (already probing the remote Kafka brokers) with two Schema Registry API-sync checks that gate both the validate_only dry run and a real create: source reachability (link_sr_unreachable) and target context emptiness (link_sr_target_not_empty, scoped to the mapped destination contexts). Both surface to the admin ConnectRPC as FAILED_PRECONDITION. --- src/v/cluster_link/BUILD | 13 + src/v/cluster_link/errc.cc | 6 + src/v/cluster_link/errc.h | 3 + src/v/cluster_link/fwd.h | 1 + src/v/cluster_link/manager.cc | 13 +- src/v/cluster_link/manager.h | 6 +- src/v/cluster_link/service.cc | 25 +- src/v/cluster_link/service.h | 12 +- src/v/cluster_link/sr_preflight_checker.cc | 290 ++++++++++++++++++ src/v/cluster_link/sr_preflight_checker.h | 80 +++++ src/v/cluster_link/tests/BUILD | 7 + src/v/cluster_link/tests/deps.cc | 4 + src/v/cluster_link/tests/deps.h | 23 ++ src/v/cluster_link/tests/link_test.cc | 256 ++++++++++++++++ .../admin/services/shadow_link/err.cc | 3 + .../admin/services/shadow_link/shadow_link.cc | 3 +- 16 files changed, 720 insertions(+), 25 deletions(-) create mode 100644 src/v/cluster_link/sr_preflight_checker.cc create mode 100644 src/v/cluster_link/sr_preflight_checker.h diff --git a/src/v/cluster_link/BUILD b/src/v/cluster_link/BUILD index 689166efd3be0..8ae784cdcc0dd 100644 --- a/src/v/cluster_link/BUILD +++ b/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", diff --git a/src/v/cluster_link/errc.cc b/src/v/cluster_link/errc.cc index b58ffcab8cfd6..d9dcfe4eeea6f 100644 --- a/src/v/cluster_link/errc.cc +++ b/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)"; diff --git a/src/v/cluster_link/errc.h b/src/v/cluster_link/errc.h index ca2a5d27545cb..f82b15268f491 100644 --- a/src/v/cluster_link/errc.h +++ b/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; diff --git a/src/v/cluster_link/fwd.h b/src/v/cluster_link/fwd.h index df4e927ee9af7..17f4b350fee12 100644 --- a/src/v/cluster_link/fwd.h +++ b/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 diff --git a/src/v/cluster_link/manager.cc b/src/v/cluster_link/manager.cc index 534a40da219ce..48d64ab452244 100644 --- a/src/v/cluster_link/manager.cc +++ b/src/v/cluster_link/manager.cc @@ -92,6 +92,7 @@ manager::manager( std::unique_ptr partition_metadata_provider, std::unique_ptr kafka_rpc_client_service, std::unique_ptr members_table_provider, + std::unique_ptr sr_preflight, ss::lowres_clock::duration task_reconciler_interval, config::binding 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 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> -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> manager::test_connection(model::metadata md) { auto err = co_await link_preflight_checks(md); if (err.code() != errc::success) { co_return err; diff --git a/src/v/cluster_link/manager.h b/src/v/cluster_link/manager.h index b244b806b06da..c4faf99f15f7b 100644 --- a/src/v/cluster_link/manager.h +++ b/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, std::unique_ptr kafka_rpc_client_service, std::unique_ptr members_table_provider, + std::unique_ptr sr_preflight, ss::lowres_clock::duration task_reconciler_interval, config::binding default_topic_replication, ss::scheduling_group scheduling_group); @@ -214,8 +216,7 @@ class manager { members_table_provider& get_members_table_provider() noexcept; - ss::future> - test_connection(model::name_t name, model::connection_config config); + ss::future> 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; std::unique_ptr _kafka_rpc_client_service; std::unique_ptr _members_table_provider; + std::unique_ptr _sr_preflight; ssx::work_queue _queue; chunked_vector> _task_factories; diff --git a/src/v/cluster_link/service.cc b/src/v/cluster_link/service.cc index 966b1cb9e59e7..71b33b5df6c34 100644 --- a/src/v/cluster_link/service.cc +++ b/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> -service::test_connection(model::name_t name, model::connection_config config) { +ss::future> 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( _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(); co_await _manager->register_task_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(); co_await _manager diff --git a/src/v/cluster_link/service.h b/src/v/cluster_link/service.h index bd2d1bc689813..8511a3f189204 100644 --- a/src/v/cluster_link/service.h +++ b/src/v/cluster_link/service.h @@ -135,16 +135,18 @@ class service : public ss::peering_sharded_service { delete_cluster_link(model::name_t name, bool force_delete_link); /** - * @brief Tests connectivity and permissions to a source cluster without + * @brief Tests connectivity and permissions for a prospective link without * persisting any state. Used to implement the validate_only path of * CreateShadowLink. * - * @param name Link name used in log and error messages - * @param config Connection configuration to test + * Takes the full metadata (not just the connection config) so that + * replication-specific preflight checks, such as the Schema Registry API + * sync checks, can inspect the link configuration. + * + * @param md Prospective link metadata to validate * @return nothing on success or a preflight error */ - ss::future> - test_connection(model::name_t name, model::connection_config config); + ss::future> test_connection(model::metadata md); /** * @brief Removes a shadow topic from a shadow link, removing all state but diff --git a/src/v/cluster_link/sr_preflight_checker.cc b/src/v/cluster_link/sr_preflight_checker.cc new file mode 100644 index 0000000000000..dcd86ca947945 --- /dev/null +++ b/src/v/cluster_link/sr_preflight_checker.cc @@ -0,0 +1,290 @@ +/* + * 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/sr_preflight_checker.h" + +#include "cluster_link/logger.h" +#include "cluster_link/model/sr_context_mapping.h" +#include "cluster_link/model/types.h" +#include "cluster_link/schema_registry_sync/http_source_reader.h" +#include "cluster_link/schema_registry_sync/source_reader.h" +#include "container/chunked_hash_map.h" +#include "container/chunked_vector.h" +#include "pandaproxy/schema_registry/types.h" +#include "schema/registry.h" + +#include +#include +#include +#include + +#include + +#include + +namespace cluster_link { + +namespace { +namespace ppsr = pandaproxy::schema_registry; + +using shadow_schema_registry_api + = model::schema_registry_sync_config::shadow_schema_registry_api; + +// Deadline for the reachability probe. The source reader's own retry budget is +// far longer than the admin request deadline, so without a shorter bound an +// unreachable source would exhaust the RPC (surfacing as a transient +// 'unavailable') instead of a prompt link_sr_unreachable. +constexpr auto sr_probe_timeout = std::chrono::seconds{5}; + +class source_sr_prober_impl : public source_sr_prober { +public: + ss::future> check_source_reachable( + const shadow_schema_registry_api& cfg, ss::abort_source& as) final { + // The factory never throws: a null or unparseable config yields an + // unavailable reader whose list_contexts reports source_unavailable, + // which we map to link_sr_unreachable below. + auto reader = schema_registry_sync::http_source_reader_factory{}.create( + &cfg); + + // Bound the probe: abort after sr_probe_timeout (or when the caller + // aborts) so an unreachable source fails fast instead of running the + // reader's full internal retry budget and timing out the admin request. + auto probe_as = ss::abort_source{}; + auto deadline = ss::timer{ + [&probe_as]() noexcept { probe_as.request_abort(); }}; + deadline.arm(sr_probe_timeout); + const auto parent_sub = as.subscribe( + [&probe_as]() noexcept { probe_as.request_abort(); }); + if (!parent_sub) { + // Caller already aborted (e.g. shard shutdown); mirror it. + probe_as.request_abort(); + } + + auto result = co_await ss::coroutine::as_future( + reader->list_contexts(probe_as)); + + // Best-effort cleanup of the throwaway probe reader; stop() closes the + // underlying rest_client, which can throw. A failure to close cleanly + // says nothing about source reachability, which we already have in + // `result`, so log and ignore it rather than let it mask the outcome. + auto stop_fut = co_await ss::coroutine::as_future(reader->stop()); + if (stop_fut.failed()) { + const auto stop_ex = stop_fut.get_exception(); + vlog( + cllog.debug, + "error stopping source schema registry reader for '{}': {}", + cfg.source_url, + stop_ex); + } + + if (result.failed()) { + const auto probe_ex = result.get_exception(); + co_return err_info( + errc::link_sr_unreachable, + fmt::format( + "failed to connect to source schema registry at '{}': {}", + cfg.source_url, + probe_ex)); + } + if (const auto contexts = result.get(); !contexts.has_value()) { + co_return err_info( + errc::link_sr_unreachable, + fmt::format( + "failed to query source schema registry at '{}': {}", + cfg.source_url, + contexts.error().message)); + } + co_return outcome::success(); + } +}; + +class sr_preflight_checker_impl : public sr_preflight_checker { +public: + sr_preflight_checker_impl( + schema::registry& destination, std::unique_ptr prober) + : _destination(destination) + , _prober(std::move(prober)) {} + + ss::future + check(const model::metadata& md, ss::abort_source& as) final { + const auto* api = md.configuration.schema_registry_sync_cfg.api_mode(); + if (api == nullptr) { + // Not a Schema Registry API-sync link (topic mode or SR sync + // disabled). + co_return err_info{errc::success}; + } + + // A disabled destination registry can be neither verified nor imported + // into; fail fast with a clear reason instead of letting its methods + // throw a logic_error below and surfacing it as an opaque failure. + if (!_destination.is_enabled()) { + co_return err_info( + errc::link_sr_verification_failed, + fmt::format( + "cannot verify schema registry for link '{}': the target " + "cluster has no schema registry enabled", + md.name)); + } + + try { + // Check #1: connectivity and authentication against the source + // Schema Registry. + const auto reachable = co_await _prober->check_source_reachable( + *api, as); + if (reachable.has_error()) { + const auto& err = reachable.assume_error(); + vlog( + cllog.warn, + "Cluster link '{}' schema registry preflight check failed - " + "{}", + md.name, + err.message()); + co_return err; + } + + // Check #2: every destination context this link would import into + // must be empty, matching Confluent IMPORT-mode semantics. Both the + // set of imported-into contexts and their emptiness are derived + // from the link config and the destination registry, not the + // source's contents. + // + // Force the local store to catch up before treating a context as + // empty: the reads below hit unsynced local shard state, so on a + // lagging replica a populated context could otherwise read empty + // and let a colliding link through. + co_await _destination.sync(); + const auto offending = co_await collect_offending_target_contexts( + *api); + if (!offending.empty()) { + vlog( + cllog.warn, + "Cluster link '{}' schema registry preflight check failed - " + "target context(s) not empty: [{}]", + md.name, + fmt::join(offending, ", ")); + co_return err_info( + errc::link_sr_target_not_empty, + fmt::format( + "target schema registry context(s) not empty: [{}]", + fmt::join(offending, ", "))); + } + } catch (...) { + const auto ex = std::current_exception(); + // A shard shutdown aborts the in-flight reads; report it as such + // rather than a verification failure the operator might retry. + if (as.abort_requested()) { + co_return err_info( + errc::service_shutting_down, + fmt::format( + "schema registry preflight for link '{}' aborted by " + "shutdown", + md.name)); + } + vlog( + cllog.warn, + "Cluster link '{}' schema registry preflight check failed - {}", + md.name, + ex); + co_return err_info( + errc::link_sr_verification_failed, + fmt::format( + "failed to verify schema registry for link '{}' - {}", + md.name, + ex)); + } + co_return err_info{errc::success}; + } + +private: + // Destination contexts the link would import into that are not empty. + // include_deleted::yes because a soft-deleted subject still occupies the + // context's namespace and would collide on import. + ss::future> + collect_offending_target_contexts(const shadow_schema_registry_api& api) { + // An unset destination mapping defaults to identity mapping. + if (!api.destination.has_value()) { + return identity_offending_contexts(api); + } + return ss::visit( + api.destination.value(), + [&]( + const model::schema_registry_sync_config::exact_context_mapping& + m) { return exact_offending_contexts(api, m.mappings); }, + [&]( + const model::schema_registry_sync_config:: + identity_context_mapping&) { + return identity_offending_contexts(api); + }); + } + + // Exact mapping: the imported-into contexts are exactly the mapping + // destinations whose source the filter selects -- a bounded set known from + // config, so probe each with has_subjects (short-circuits). A mapping whose + // source the filter excludes is inert, so requiring its destination to be + // empty would over-block. + ss::future> exact_offending_contexts( + const shadow_schema_registry_api& api, + const chunked_hash_map& mappings) { + auto targets = mappings | std::views::filter([&](const auto& kv) { + return model::filter_selects_source_context( + api.filter, ppsr::context{kv.first}); + }) + | std::views::values + | std::ranges::to>(); + + auto offending = chunked_vector{}; + for (const auto& ctx : targets) { + if ( + co_await _destination.has_subjects( + ppsr::context{ctx}, ppsr::include_deleted::yes)) { + offending.push_back(ctx); + } + } + co_return offending; + } + + // Identity/unset mapping: identity keeps the context name, so a destination + // context is imported into iff the link filter selects it. Any in-scope + // destination subject therefore makes its context offending. This mirrors + // the runtime client-write blocker in frontend.cc + // (api_mode_shadows_context), which likewise keys identity ownership off + // the destination context and the filter. + ss::future> + identity_offending_contexts(const shadow_schema_registry_api& api) { + auto subjects = co_await _destination.get_subjects( + ppsr::include_deleted::yes); + + co_return subjects + | std::views::filter([&api](const ppsr::context_subject& cs) { + return model::filter_selects_source_context(api.filter, cs.ctx); + }) + | std::views::transform([](const auto& cs) { return cs.ctx(); }) + | std::ranges::to>() + | std::ranges::to>(); + } + + schema::registry& _destination; + std::unique_ptr _prober; +}; + +} // namespace + +std::unique_ptr source_sr_prober::make_default() { + return std::make_unique(); +} + +std::unique_ptr sr_preflight_checker::make_default( + schema::registry& destination, std::unique_ptr prober) { + return std::make_unique( + destination, std::move(prober)); +} + +} // namespace cluster_link diff --git a/src/v/cluster_link/sr_preflight_checker.h b/src/v/cluster_link/sr_preflight_checker.h new file mode 100644 index 0000000000000..dc17c5749f316 --- /dev/null +++ b/src/v/cluster_link/sr_preflight_checker.h @@ -0,0 +1,80 @@ +/* + * 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/errc.h" +#include "cluster_link/model/types.h" +#include "schema/fwd.h" + +#include +#include + +#include + +namespace cluster_link { + +/// \brief Probes a prospective link's source Schema Registry during preflight. +/// +/// Builds a Schema Registry REST client from the API-mode source connection via +/// `schema_registry_sync::make_source_sr_client` and issues a list to confirm +/// the source is reachable with the supplied credentials. +/// +/// The shared `make_source_sr_client` builder is intended to also back the +/// real HTTP-backed `source_reader` +/// (cluster_link/schema_registry_sync/source_reader.h, today only +/// `unavailable_source_reader`) once it is wired, so the preflight and the sync +/// read path construct the source-SR client one way. +class source_sr_prober { +public: + source_sr_prober() = default; + source_sr_prober(const source_sr_prober&) = delete; + source_sr_prober(source_sr_prober&&) = delete; + source_sr_prober& operator=(const source_sr_prober&) = delete; + source_sr_prober& operator=(source_sr_prober&&) = delete; + virtual ~source_sr_prober() = default; + + static std::unique_ptr make_default(); + + /// Connects to the source Schema Registry described by \p cfg to confirm it + /// is reachable with the supplied credentials. A failed connection or + /// authentication is reported as an `errc::link_sr_unreachable` error. + virtual ss::future> check_source_reachable( + const model::schema_registry_sync_config::shadow_schema_registry_api& cfg, + ss::abort_source& as) = 0; +}; + +/// \brief Validates the Schema Registry side of a prospective cluster link. +/// +/// Owns the collaborators the check needs (a `source_sr_prober` for the source +/// side and a reference to the local destination `schema::registry`) so that +/// callers only depend on this one seam rather than on the individual pieces. +class sr_preflight_checker { +public: + sr_preflight_checker() = default; + sr_preflight_checker(const sr_preflight_checker&) = delete; + sr_preflight_checker(sr_preflight_checker&&) = delete; + sr_preflight_checker& operator=(const sr_preflight_checker&) = delete; + sr_preflight_checker& operator=(sr_preflight_checker&&) = delete; + virtual ~sr_preflight_checker() = default; + + static std::unique_ptr make_default( + schema::registry& destination, std::unique_ptr); + + /// Validates a prospective link's Schema Registry API-sync configuration: + /// that the source Schema Registry is reachable with the supplied + /// credentials, and that every destination context the link would import + /// into is empty. A no-op for links not using Schema Registry API mode. + virtual ss::future + check(const model::metadata& md, ss::abort_source& as) = 0; +}; + +} // namespace cluster_link diff --git a/src/v/cluster_link/tests/BUILD b/src/v/cluster_link/tests/BUILD index 1f67b04c2f486..f3bb0b156ad8d 100644 --- a/src/v/cluster_link/tests/BUILD +++ b/src/v/cluster_link/tests/BUILD @@ -24,6 +24,8 @@ redpanda_test_cc_library( "//src/v/kafka/data/rpc", "//src/v/kafka/data/rpc/test:test_deps", "//src/v/model", + "//src/v/pandaproxy/schema_registry:types", + "//src/v/schema/tests:fake_registry", "//src/v/security", "@seastar", ], @@ -41,7 +43,12 @@ redpanda_cc_gtest( "//src/v/cluster/cluster_link/tests:test_lib", "//src/v/cluster_link:impl", "//src/v/cluster_link/replication/tests:deps_test_impl", + "//src/v/kafka/protocol:find_coordinator", + "//src/v/kafka/protocol:list_groups", + "//src/v/kafka/protocol:list_offset", + "//src/v/kafka/protocol:offset_fetch", "//src/v/model", + "//src/v/schema/tests:fake_registry", "//src/v/test_utils:gtest", "@abseil-cpp//absl/container:flat_hash_map", "@googletest//:gtest", diff --git a/src/v/cluster_link/tests/deps.cc b/src/v/cluster_link/tests/deps.cc index 7fb0ff364f4b2..3ac2eff2fe616 100644 --- a/src/v/cluster_link/tests/deps.cc +++ b/src/v/cluster_link/tests/deps.cc @@ -101,6 +101,10 @@ ss::future<> cluster_link_manager_test_fixture::wire_up_and_start( _fmtp = fmtp.get(); return fmtp; }), + ss::sharded_parameter([this]() { + return sr_preflight_checker::make_default( + _fake_schema_registry, std::make_unique()); + }), 1s, _default_topic_replication.bind(), ss::default_scheduling_group()); diff --git a/src/v/cluster_link/tests/deps.h b/src/v/cluster_link/tests/deps.h index 60a91da265958..fa936d207bf5c 100644 --- a/src/v/cluster_link/tests/deps.h +++ b/src/v/cluster_link/tests/deps.h @@ -22,6 +22,7 @@ #include "kafka/client/test/cluster_mock.h" #include "kafka/data/rpc/deps.h" #include "kafka/data/rpc/test/deps.h" +#include "schema/tests/fake_registry.h" #include "security/acl_entry_set.h" #include "security/role.h" #include "security/role_store.h" @@ -36,6 +37,27 @@ using data_sink_factory = cluster_link::replication::tests::accounting_sink_factory; namespace cluster_link::tests { +/// Source Schema Registry prober test double. Reachable by default, or reports +/// an error when one is set. +class fake_source_sr_prober : public source_sr_prober { +public: + ss::future> check_source_reachable( + const model::schema_registry_sync_config::shadow_schema_registry_api&, + ss::abort_source&) override { + ++call_count; + if (error.has_value()) { + return ss::make_ready_future>(error.value()); + } + return ss::make_ready_future>(outcome::success()); + } + + std::optional error; + // Number of times check_source_reachable was invoked; lets tests assert the + // source probe ran (or was skipped) rather than inferring it from the + // preflight result alone. + size_t call_count{0}; +}; + class test_link_factory : public link_factory { public: explicit test_link_factory( @@ -793,6 +815,7 @@ class cluster_link_manager_test_fixture { test_partition_metadata_provider* _partition_metadata_provider{nullptr}; test_kafka_rpc_client_service* _tkrcs{nullptr}; fake_members_table_provider* _fmtp{nullptr}; + schema::fake_registry _fake_schema_registry; ss::sharded _manager; config::mock_property _default_topic_replication{1}; diff --git a/src/v/cluster_link/tests/link_test.cc b/src/v/cluster_link/tests/link_test.cc index a53bc71533b63..676c16f9a6519 100644 --- a/src/v/cluster_link/tests/link_test.cc +++ b/src/v/cluster_link/tests/link_test.cc @@ -16,10 +16,16 @@ #include "cluster_link/manager.h" #include "cluster_link/replication/tests/deps_test_impl.h" #include "cluster_link/tests/deps.h" +#include "kafka/protocol/find_coordinator.h" +#include "kafka/protocol/list_groups.h" +#include "kafka/protocol/list_offset.h" +#include "kafka/protocol/offset_fetch.h" +#include "schema/tests/fake_registry.h" #include "test_utils/test.h" #include +#include #include using namespace std::chrono_literals; @@ -146,6 +152,7 @@ class link_test_base : public seastar_test { ss::sharded _table; std::unique_ptr _manager; + schema::fake_registry _fake_schema_registry; config::mock_property _default_topic_replication{3}; absl::flat_hash_map> @@ -153,6 +160,12 @@ class link_test_base : public seastar_test { notification_id _latest_id{0}; private: + template + void advertise_api() { + _cluster_mock.default_supported_versions[Api::key] = { + .min = Api::min_valid, .max = Api::max_valid}; + } + void setup_cluster_mock() { _cluster_mock.register_default_handlers(); _cluster_mock.add_broker( @@ -161,6 +174,14 @@ class link_test_base : public seastar_test { ::model::node_id(1), net::unresolved_address{"localhost", 9093}); _cluster_mock.add_broker( ::model::node_id(2), net::unresolved_address{"localhost", 9094}); + // Advertise (via the mock's default fallback) the APIs the link broker + // preflight requires that are not in the mock's built-in defaults, so + // the kafka-path check passes and the Schema Registry checks are + // reached. + advertise_api(); + advertise_api(); + advertise_api(); + advertise_api(); } }; @@ -171,6 +192,8 @@ class link_test : public link_test_base { co_await link_test_base::SetUpAsync(); auto tmc = std::make_unique(); _tmc = tmc.get(); + auto sr_prober = std::make_unique(); + _fake_sr_prober = sr_prober.get(); _manager = std::make_unique( ::model::node_id(0), std::make_unique( @@ -194,6 +217,8 @@ class link_test : public link_test_base { std::make_unique(), std::make_unique(_tmc), std::make_unique(), + sr_preflight_checker::make_default( + _fake_schema_registry, std::move(sr_prober)), task_reconciler_interval, _default_topic_replication.bind(), ss::default_scheduling_group()); @@ -218,6 +243,7 @@ class link_test : public link_test_base { protected: absl::flat_hash_map _links; fake_topic_metadata_cache* _tmc{nullptr}; + fake_source_sr_prober* _fake_sr_prober{nullptr}; }; class link_test_manager_started : public link_test { @@ -440,6 +466,8 @@ class evil_link_test : public link_test_base { std::make_unique(), std::make_unique(_tmc), std::make_unique(), + sr_preflight_checker::make_default( + _fake_schema_registry, std::make_unique()), task_reconciler_interval, _default_topic_replication.bind(), ss::default_scheduling_group()); @@ -595,4 +623,232 @@ TEST_F_CORO( co_await wedged_link->stop(); } +// --- Schema Registry API-sync preflight checks --- + +namespace { +namespace ppsr = pandaproxy::schema_registry; +using sr_cfg = model::schema_registry_sync_config; + +// Matches a cl_result that is an error carrying the given errc. Reports +// the actual state (value, or the mismatching code) on failure. Usable inside +// coroutine tests because EXPECT_THAT is non-fatal and never returns. +MATCHER_P(IsLinkError, expected_code, "") { + if (arg.has_value()) { + *result_listener << "is a success, expected error"; + return false; + } + *result_listener << "error code " + << static_cast(arg.assume_error().code()) << " (" + << arg.assume_error().message() << ")"; + return arg.assume_error().code() == expected_code; +} + +// Builds link metadata whose kafka connection targets the fixture's broker +// mock (so the kafka preflight passes) and whose Schema Registry sync config is +// in API mode with the given destination mapping and source context filter. +model::metadata make_api_sr_metadata( + ss::sstring name, + std::optional destination, + chunked_vector filter_contexts = {}, + chunked_vector filter_subjects = {}) { + model::metadata md{ + .name = model::name_t(std::move(name)), + .uuid = model::uuid_t(::uuid_t::create()), + .connection = model::connection_config{ + .bootstrap_servers{net::unresolved_address{"localhost", 9092}}}}; + sr_cfg::shadow_schema_registry_api api; + api.source_url = "http://source.example:8081"; + api.filter.contexts = std::move(filter_contexts); + api.filter.subjects = std::move(filter_subjects); + api.destination = std::move(destination); + md.configuration.schema_registry_sync_cfg.sync_mode = std::move(api); + return md; +} + +sr_cfg::destination_mapping_t identity_mapping() { + return sr_cfg::destination_mapping_t{sr_cfg::identity_context_mapping{}}; +} + +sr_cfg::destination_mapping_t +exact_mapping(ss::sstring source, ss::sstring destination) { + sr_cfg::exact_context_mapping mapping; + mapping.mappings.emplace(std::move(source), std::move(destination)); + return sr_cfg::destination_mapping_t{std::move(mapping)}; +} + +ss::future<> seed_subject( + schema::fake_registry& reg, + ppsr::context ctx, + ss::sstring sub, + ppsr::is_deleted deleted = ppsr::is_deleted::no) { + co_await reg.import_schema( + ppsr::stored_schema{ + .schema = ppsr:: + subject_schema{ppsr::context_subject{std::move(ctx), ppsr::subject{std::move(sub)}}, ppsr::schema_definition{ppsr::schema_definition::raw_string{R"({"type":"string"})"}, ppsr::schema_type::avro}}, + .version = ppsr::schema_version{1}, + .id = ppsr::schema_id{1}, + .deleted = deleted}); +} +} // namespace + +// A link that is not in Schema Registry API mode skips both SR checks: the +// source prober is not consulted even though it is primed to fail. +TEST_F_CORO(link_test_manager_started, sr_preflight_skipped_when_not_api_mode) { + _fake_sr_prober->error = err_info( + errc::link_sr_unreachable, "prober must not be consulted"); + model::metadata md{ + .name = model::name_t("link1"), + .uuid = model::uuid_t(::uuid_t::create()), + .connection = model::connection_config{ + .bootstrap_servers{net::unresolved_address{"localhost", 9092}}}}; + auto res = co_await _manager->test_connection(std::move(md)); + ASSERT_TRUE_CORO(res.has_value()); + EXPECT_THAT(_fake_sr_prober->call_count, testing::Eq(0)); +} + +// Source reachable and the destination empty: preflight passes, and the source +// probe actually ran. +TEST_F_CORO(link_test_manager_started, sr_preflight_reachable_and_empty_ok) { + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping())); + ASSERT_TRUE_CORO(res.has_value()); + EXPECT_THAT(_fake_sr_prober->call_count, testing::Eq(1)); +} + +// The prober reporting an error surfaces as link_sr_unreachable. +TEST_F_CORO(link_test_manager_started, sr_preflight_source_unreachable) { + _fake_sr_prober->error = err_info( + errc::link_sr_unreachable, "connection refused"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping())); + EXPECT_THAT(res, IsLinkError(errc::link_sr_unreachable)); +} + +// Identity mapping, select-all filter: a populated destination context fails. +TEST_F_CORO(link_test_manager_started, sr_preflight_identity_target_not_empty) { + co_await seed_subject( + _fake_schema_registry, ppsr::context{".ctxA"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping())); + EXPECT_THAT(res, IsLinkError(errc::link_sr_target_not_empty)); +} + +// Identity mapping: a populated in-scope destination context fails even when +// the source has no such context. Only the destination registry is inspected +// for emptiness (the prober reports nothing here), so a context present only in +// the destination is still caught. +TEST_F_CORO( + link_test_manager_started, sr_preflight_identity_target_absent_from_source) { + co_await seed_subject( + _fake_schema_registry, ppsr::context{".ctxA"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping(), {".ctxA"})); + EXPECT_THAT(res, IsLinkError(errc::link_sr_target_not_empty)); +} + +// Exact mapping keys emptiness off the destination context name, not the +// source: a subject in the source-named context does not fail the check. +TEST_F_CORO( + link_test_manager_started, sr_preflight_exact_mapping_checks_destination) { + co_await seed_subject(_fake_schema_registry, ppsr::context{".src"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", exact_mapping(".src", ".dst"))); + ASSERT_TRUE_CORO(res.has_value()); + EXPECT_THAT(_fake_sr_prober->call_count, testing::Eq(1)); +} + +// Exact mapping: a subject in the destination context fails the check. +TEST_F_CORO( + link_test_manager_started, sr_preflight_exact_mapping_destination_not_empty) { + co_await seed_subject(_fake_schema_registry, ppsr::context{".dst"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", exact_mapping(".src", ".dst"))); + EXPECT_THAT(res, IsLinkError(errc::link_sr_target_not_empty)); +} + +// The context filter narrows the identity target set: a populated context +// excluded by the filter is out of scope. +TEST_F_CORO( + link_test_manager_started, sr_preflight_identity_filtered_context_ignored) { + co_await seed_subject( + _fake_schema_registry, ppsr::context{".ctxB"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping(), {".ctxA"})); + ASSERT_TRUE_CORO(res.has_value()); + EXPECT_THAT(_fake_sr_prober->call_count, testing::Eq(1)); +} + +// A subjects-only filter selects the source context parsed from the qualified +// subject, so identity mapping imports into it and its populated target is +// checked. Guards against dropping filter.subjects from target resolution. +TEST_F_CORO( + link_test_manager_started, sr_preflight_identity_subject_filter_selects) { + co_await seed_subject( + _fake_schema_registry, ppsr::context{".ctxB"}, "sub1"); + // Qualified subject ":.ctxB:sub1" selects source context ".ctxB". + auto res = co_await _manager->test_connection(make_api_sr_metadata( + "link1", identity_mapping(), /*filter_contexts=*/{}, {":.ctxB:sub1"})); + EXPECT_THAT(res, IsLinkError(errc::link_sr_target_not_empty)); +} + +// A context selected by neither filter.contexts nor filter.subjects is out of +// scope, even when populated. +TEST_F_CORO( + link_test_manager_started, sr_preflight_identity_subject_filter_excludes) { + co_await seed_subject( + _fake_schema_registry, ppsr::context{".ctxB"}, "sub1"); + auto res = co_await _manager->test_connection(make_api_sr_metadata( + "link1", identity_mapping(), /*filter_contexts=*/{}, {":.ctxA:sub1"})); + ASSERT_TRUE_CORO(res.has_value()); + EXPECT_THAT(_fake_sr_prober->call_count, testing::Eq(1)); +} + +// Exact mapping whose source is excluded by the filter is inert, so its +// populated destination context does not fail the check. +TEST_F_CORO( + link_test_manager_started, sr_preflight_exact_mapping_filtered_source_inert) { + co_await seed_subject(_fake_schema_registry, ppsr::context{".dst"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", exact_mapping(".src", ".dst"), {".other"})); + ASSERT_TRUE_CORO(res.has_value()); + EXPECT_THAT(_fake_sr_prober->call_count, testing::Eq(1)); +} + +// A soft-deleted subject still occupies the context namespace and counts as +// non-empty (include_deleted::yes). +TEST_F_CORO( + link_test_manager_started, sr_preflight_soft_deleted_counts_as_not_empty) { + co_await seed_subject( + _fake_schema_registry, + ppsr::context{".ctxA"}, + "sub1", + ppsr::is_deleted::yes); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping())); + EXPECT_THAT(res, IsLinkError(errc::link_sr_target_not_empty)); +} + +// An unset destination mapping defaults to identity mapping. +TEST_F_CORO( + link_test_manager_started, + sr_preflight_unset_destination_defaults_to_identity) { + co_await seed_subject( + _fake_schema_registry, ppsr::context{".ctxA"}, "sub1"); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", std::nullopt)); + EXPECT_THAT(res, IsLinkError(errc::link_sr_target_not_empty)); +} + +// A fault querying the destination registry surfaces as +// link_sr_verification_failed. +TEST_F_CORO( + link_test_manager_started, + sr_preflight_destination_fault_verification_failed) { + _fake_schema_registry.set_inject_failures( + std::make_exception_ptr(std::runtime_error("registry unavailable"))); + auto res = co_await _manager->test_connection( + make_api_sr_metadata("link1", identity_mapping())); + EXPECT_THAT(res, IsLinkError(errc::link_sr_verification_failed)); +} + } // namespace cluster_link::tests diff --git a/src/v/redpanda/admin/services/shadow_link/err.cc b/src/v/redpanda/admin/services/shadow_link/err.cc index 00eb82877e86f..3c48d881694eb 100644 --- a/src/v/redpanda/admin/services/shadow_link/err.cc +++ b/src/v/redpanda/admin/services/shadow_link/err.cc @@ -46,6 +46,9 @@ void handle_error(cluster_link::errc err, ss::sstring info) { case cluster_link::errc::link_broker_unreachable: case cluster_link::errc::link_broker_verification_failed: case cluster_link::errc::link_verification_unknown_error: + case cluster_link::errc::link_sr_unreachable: + case cluster_link::errc::link_sr_target_not_empty: + case cluster_link::errc::link_sr_verification_failed: throw serde::pb::rpc::failed_precondition_exception(std::move(info)); case cluster_link::errc::link_id_not_found: case cluster_link::errc::topic_not_being_mirrored: diff --git a/src/v/redpanda/admin/services/shadow_link/shadow_link.cc b/src/v/redpanda/admin/services/shadow_link/shadow_link.cc index fcd39ddbb103f..735575b5e29e4 100644 --- a/src/v/redpanda/admin/services/shadow_link/shadow_link.cc +++ b/src/v/redpanda/admin/services/shadow_link/shadow_link.cc @@ -74,8 +74,7 @@ shadow_link_service_impl::create_shadow_link( } if (validate_only) { - handle_error( - co_await _service->local().test_connection(md.name, md.connection)); + handle_error(co_await _service->local().test_connection(std::move(md))); co_return proto::admin::create_shadow_link_response{}; } From 51f1e0706649922213c9b9fd008332066c66a44d Mon Sep 17 00:00:00 2001 From: Mateusz Najda Date: Fri, 3 Jul 2026 09:10:30 +0200 Subject: [PATCH 5/5] ducktape: end-to-end shadow-link SR preflight ducktape Two-cluster coverage of the SR preflight over validate_only and real create, positive and negative, asserting the specific failure message. --- .../tests/shadow_link_sr_preflight_test.py | 193 ++++++++++++++++++ .../type-checking/type-check-strictness.json | 1 + 2 files changed, 194 insertions(+) create mode 100644 tests/rptest/tests/shadow_link_sr_preflight_test.py diff --git a/tests/rptest/tests/shadow_link_sr_preflight_test.py b/tests/rptest/tests/shadow_link_sr_preflight_test.py new file mode 100644 index 0000000000000..ab13334856226 --- /dev/null +++ b/tests/rptest/tests/shadow_link_sr_preflight_test.py @@ -0,0 +1,193 @@ +# 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 + +from typing import Any + +from connectrpc.errors import ConnectError, ConnectErrorCode +from ducktape.utils.util import wait_until + +from rptest.clients.admin.proto.redpanda.core.admin.v2 import shadow_link_pb2 +from rptest.clients.rpk import RpkTool +from rptest.services.cluster import TestContext, cluster +from rptest.services.multi_cluster_services import SecondaryClusterArgs +from rptest.services.redpanda import SchemaRegistryConfig +from rptest.tests.cluster_linking_test_base import ShadowLinkTestBase +from rptest.util import expect_exception + +_SCHEMA = ( + '{"type":"record","name":"preflight_record",' + '"fields":[{"name":"f1","type":"string"}]}' +) + + +class ShadowLinkSchemaRegistryPreflightTest(ShadowLinkTestBase): + """End-to-end coverage for the Schema Registry API-sync + preflight checks that run on shadow-link creation (including the + validate_only pre-flight): source Schema Registry reachability and target + context emptiness. Both clusters run Schema Registry so the source is a real + reachable endpoint and the target registry is queryable.""" + + LINK_NAME = "sr-preflight-link" + + def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any): + super().__init__( + test_context=test_context, + num_brokers=3, + secondary_cluster_args=SecondaryClusterArgs( + schema_registry_config=SchemaRegistryConfig() + ), + schema_registry_config=SchemaRegistryConfig(), + *args, + **kwargs, + ) + + def _source_sr_url(self) -> str: + return self.source_cluster_service.schema_reg(limit=1) + + def _validate_request( + self, + source_url: str, + destination: shadow_link_pb2.SchemaRegistryContextDestination | None = None, + validate_only: bool = True, + ) -> shadow_link_pb2.CreateShadowLinkRequest: + req = self.create_default_link_request( + link_name=self.LINK_NAME, + mirror_all_acls=False, + mirror_all_groups=False, + mirror_all_topics=False, + ) + api = shadow_link_pb2.SchemaRegistrySyncOptions.ShadowSchemaRegistryApi( + source_url=source_url + ) + if destination is not None: + api.destination.CopyFrom(destination) + req.shadow_link.configurations.schema_registry_sync_options.shadow_schema_registry_api.CopyFrom( + api + ) + # validate_only runs the preflight without creating the link; when False + # the same checks gate a real create. + req.validate_only = validate_only + return req + + def _seed_subject(self, rpk: RpkTool, subject: str) -> None: + rpk.create_schema_from_str(subject, _SCHEMA) + + def setUp(self): + super().setUp() + # The shadow_link_sr_api_sync feature auto-activates once every broker + # is at least v26.2, which holds for this cluster, so no manual + # activation is needed for the request to reach the preflight checks. + # + # Wait for the shadow-link admin service to be ready so the first + # create/validate request does not race a still-initializing controller + # and return a transient 'unavailable'. + wait_until( + lambda: self.list_links() is not None, + timeout_sec=30, + backoff_sec=1, + retry_on_exc=True, + err_msg="shadow link admin service did not become ready", + ) + + @cluster(num_nodes=6) # 3 target + 3 source brokers + def test_validate_reachable_empty_target_ok(self): + """Source reachable and the (identity-mapped) target context empty: + validation succeeds.""" + req = self._validate_request(self._source_sr_url()) + # Raises ConnectError on failure; success returns normally. + self.create_link_with_request(req=req) + + # validate_only must not have created the link. + assert self.LINK_NAME not in [link.name for link in self.list_links()], ( + "validate_only unexpectedly created the link" + ) + + @cluster(num_nodes=6) + def test_validate_source_unreachable(self): + """An unreachable source Schema Registry URL fails preflight.""" + req = self._validate_request("http://schema-registry.invalid:8081") + with expect_exception( + ConnectError, + lambda e: e.code == ConnectErrorCode.FAILED_PRECONDITION + and "source schema registry" in str(e), + ): + self.create_link_with_request(req=req) + + @cluster(num_nodes=6) + def test_validate_target_context_not_empty(self): + """A populated target context fails preflight. Emptiness is checked + against the target registry, so a subject present only on the target + (the source is reachable but empty) is enough to trip the check.""" + self._seed_subject(self.target_cluster_rpk, "target-only-value") + + req = self._validate_request(self._source_sr_url()) + with expect_exception( + ConnectError, + lambda e: e.code == ConnectErrorCode.FAILED_PRECONDITION + and "context(s) not empty" in str(e), + ): + self.create_link_with_request(req=req) + + @cluster(num_nodes=6) + def test_create_blocked_when_target_context_not_empty(self): + """The same preflight gates a real create (validate_only=False), not + just the dry run: a populated target context must reject the create and + leave no link behind.""" + self._seed_subject(self.target_cluster_rpk, "target-only-value") + + req = self._validate_request(self._source_sr_url(), validate_only=False) + with expect_exception( + ConnectError, + lambda e: e.code == ConnectErrorCode.FAILED_PRECONDITION + and "context(s) not empty" in str(e), + ): + self.create_link_with_request(req=req) + + # A rejected create must not leave a partially-created link behind. + assert self.LINK_NAME not in [link.name for link in self.list_links()], ( + "preflight-rejected create unexpectedly left the link behind" + ) + + @cluster(num_nodes=6) + def test_create_succeeds_when_reachable_and_target_empty(self): + """The positive real-create path: with a reachable source and an empty + target context, a create (validate_only=False) passes preflight and the + link is actually persisted. Guards against preflight over-blocking a + valid SR-API link on the create path.""" + req = self._validate_request(self._source_sr_url(), validate_only=False) + # Raises ConnectError on failure; success returns normally. + self.create_link_with_request(req=req) + + assert self.LINK_NAME in [link.name for link in self.list_links()], ( + "create passed preflight but the link was not persisted" + ) + + @cluster(num_nodes=6) + def test_validate_exact_mapping_checks_destination_context(self): + """Exact mapping keys emptiness off the destination context: a subject + in the target's default context fails when it is a mapping + destination.""" + self._seed_subject(self.target_cluster_rpk, "dest-value") + + destination = shadow_link_pb2.SchemaRegistryContextDestination( + exact=shadow_link_pb2.SchemaRegistryExactContextMappings( + mappings=[ + shadow_link_pb2.SchemaRegistryContextMap( + source=".source-ctx", destination="." + ) + ] + ) + ) + req = self._validate_request(self._source_sr_url(), destination=destination) + with expect_exception( + ConnectError, + lambda e: e.code == ConnectErrorCode.FAILED_PRECONDITION + and "context(s) not empty" in str(e), + ): + self.create_link_with_request(req=req) diff --git a/tools/type-checking/type-check-strictness.json b/tools/type-checking/type-check-strictness.json index d28738bb831f7..b44409e3447c7 100644 --- a/tools/type-checking/type-check-strictness.json +++ b/tools/type-checking/type-check-strictness.json @@ -199,6 +199,7 @@ "rptest/tests/security_report_test.py", "rptest/tests/services_self_test.py", "rptest/tests/shadow_indexing_firewall_test.py", + "rptest/tests/shadow_link_sr_preflight_test.py", "rptest/tests/storage_failure_injection_test.py", "rptest/tests/strict_data_init_test.py", "rptest/tests/tiered_storage_enable_test.py",