[CORE-16608] cluster_link: validate source SR reachability and target emptiness on link creation#31011
Conversation
ff5d59a to
c6cad96
Compare
There was a problem hiding this comment.
Pull request overview
Adds Schema Registry (SR) API-sync preflight validation to cluster-link creation so that both validate_only and real creates are gated on (1) source SR reachability/auth and (2) destination-context emptiness for the contexts that would be imported into. This extends existing link preflight checks (Kafka broker probing) and introduces new SR client construction utilities plus unit/integration coverage.
Changes:
- Introduces SR preflight checker (
sr_preflight_checker) and wires it intocluster_link::manager::link_preflight_checks, surfacing failures asFAILED_PRECONDITION. - Adds a shared SR REST client builder (
schema_registry_sync::make_source_sr_client) + context-filter helper (filter_selects_source_context) to keep preflight logic consistent with runtime SR behavior. - Adds new C++ unit tests + ducktape end-to-end tests for SR preflight scenarios (unreachable source, non-empty target, mapping/filter scoping).
Reviewed changes
Copilot reviewed 37 out of 37 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| tools/type-checking/type-check-strictness.json | Adds new ducktape test to strict type-check set. |
| tests/rptest/tests/shadow_link_sr_preflight_test.py | E2E ducktape coverage for SR reachability + target emptiness checks on validate/create. |
| src/v/schema/tests/fake_registry.h | Extends fake registry interface with has_subjects. |
| src/v/schema/tests/fake_registry.cc | Implements has_subjects for fake registry store. |
| src/v/schema/registry.h | Adds registry::has_subjects API for efficient emptiness checks. |
| src/v/schema/registry.cc | Implements has_subjects in enabled/disabled schema registry wrappers. |
| src/v/redpanda/admin/services/shadow_link/shadow_link.cc | Updates validate-only path to pass full metadata into test_connection. |
| src/v/redpanda/admin/services/shadow_link/err.cc | Maps new SR preflight errors to FAILED_PRECONDITION. |
| src/v/pandaproxy/schema_registry/rest_client/BUILD | Expands visibility so cluster_link can depend on SR REST client code. |
| src/v/datalake/tests/record_schema_resolver_test.cc | Updates test registry wrapper to implement has_subjects. |
| src/v/cluster/cluster_link/frontend.cc | Reuses shared context-filter predicate from cluster_link::model. |
| src/v/cluster/BUILD | Adds Bazel dep on new sr_context_mapping lib. |
| src/v/cluster_link/utils.h | Exposes TLS config builder used by both Kafka and SR client paths. |
| src/v/cluster_link/utils.cc | Refactors TLS config creation to support SR client construction. |
| src/v/cluster_link/tests/link_test.cc | Adds unit tests for SR preflight behavior and required Kafka API advertisement in mocks. |
| src/v/cluster_link/tests/deps.h | Adds fake_source_sr_prober and test fixture SR registry state. |
| src/v/cluster_link/tests/deps.cc | Wires SR preflight checker into manager fixture construction. |
| src/v/cluster_link/tests/BUILD | Adds deps for SR types + fake registry and additional Kafka protocol APIs. |
| src/v/cluster_link/sr_preflight_checker.h | Defines SR preflight checker and source SR prober interfaces. |
| src/v/cluster_link/sr_preflight_checker.cc | Implements SR reachability probe + destination-context emptiness validation. |
| src/v/cluster_link/service.h | Changes test_connection to accept full model::metadata for richer preflight. |
| src/v/cluster_link/service.cc | Constructs destination schema registry earlier and injects SR preflight checker into manager. |
| src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h | Updates delegating registry fixture to implement has_subjects. |
| src/v/cluster_link/schema_registry_sync/tests/source_client_test.cc | Unit tests for URL/TLS/auth parameter derivation for SR source client. |
| src/v/cluster_link/schema_registry_sync/tests/BUILD | Adds Bazel target for new source_client_test. |
| src/v/cluster_link/schema_registry_sync/source_client.h | Adds SR source client parameter resolution + client factory API. |
| src/v/cluster_link/schema_registry_sync/source_client.cc | Implements SR source URL parsing, TLS/SNI resolution, and client construction. |
| src/v/cluster_link/schema_registry_sync/BUILD | Adds Bazel library target for source_client. |
| src/v/cluster_link/model/sr_context_mapping.h | Declares shared predicate for SR context filter membership. |
| src/v/cluster_link/model/sr_context_mapping.cc | Implements shared predicate used by runtime and preflight. |
| src/v/cluster_link/model/BUILD | Adds Bazel library target for sr_context_mapping. |
| src/v/cluster_link/manager.h | Injects SR preflight checker dependency; updates test_connection signature. |
| src/v/cluster_link/manager.cc | Runs SR preflight after Kafka preflight and updates validate-only logic to use metadata. |
| src/v/cluster_link/fwd.h | Adds forward declaration for sr_preflight_checker. |
| src/v/cluster_link/errc.h | Adds new SR-related error codes. |
| src/v/cluster_link/errc.cc | Adds messages for new SR-related error codes. |
| src/v/cluster_link/BUILD | Adds new sources and deps for SR preflight and SR client construction. |
| uint16_t resolve_port(const ada::url_aggregator& url) { | ||
| if (url.has_port()) { | ||
| auto parsed_port = int32_t{0}; | ||
| std::ignore = absl::SimpleAtoi(url.get_port(), &parsed_port); | ||
| return static_cast<uint16_t>(parsed_port); | ||
| } | ||
| return url.type == ada::scheme::HTTPS ? sr_default_https_port | ||
| : sr_default_http_port; | ||
| } |
| const auto& url = *parsed; | ||
| const bool needs_tls = url.type == ada::scheme::HTTPS | ||
| || bool(cfg.tls_enabled); | ||
| auto host = ss::sstring(url.get_hostname()); | ||
|
|
||
| std::optional<ss::sstring> sni; | ||
| if (needs_tls && bool(cfg.tls_provide_sni)) { | ||
| sni = host; | ||
| } |
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.
Splits source-SR client construction into a pure, unit-testable resolve_source_client_params (URL/port/TLS/SNI/credentials derivation) and the async make_source_sr_client. Extracts the TLS-config builder in utils so it can be shared with the Kafka client path.
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.
c6cad96 to
931d73a
Compare
|
force-pushed to resolve a merge conflict |
bartoszpiekny-redpanda
left a comment
There was a problem hiding this comment.
Code looks great but I'm not fully familiar with the functionality so please wait for someone else accepting it 🙂
| @@ -0,0 +1,72 @@ | |||
| /* | |||
There was a problem hiding this comment.
I've skipped reviewing this commit, because I think we'll be able to just drop in the source reader client from #30999 instead of this one.
| @@ -0,0 +1,30 @@ | |||
| /* | |||
There was a problem hiding this comment.
Nice! Thanks for splitting this into a separate commit 🚀
| # The API-sync path is feature-gated; ensure it is active so the | ||
| # request reaches the preflight checks rather than the feature gate. | ||
| self.target_cluster_service.set_feature_active(SR_API_SYNC_FEATURE, True) |
There was a problem hiding this comment.
I don't think this bit is necessary here.
The way these features work is that they auto-activate when all of the brokers are at least as up to date as a specific broker version. For this api sync feature specifically, the feature auto-activates when the brokers are all on at least version 26.2.
Since this tests creates a cluster where all brokers are at least 26.2, there is no need to manually enable the feature, it will just be auto-enabled while the cluster starts up.
| // Identity/unset mapping: the in-scope source contexts (those the link's | ||
| // filter selects) become the destination contexts. Filter membership is | ||
| // resolved through the shared predicate so this stays consistent with the | ||
| // runtime client-write blocker (cluster/cluster_link/frontend.cc). | ||
| auto identity_targets = [&] { | ||
| return source_contexts | std::views::filter([&](const auto& ctx) { | ||
| return model::filter_selects_source_context(api.filter, ctx); | ||
| }) | ||
| | std::views::transform([](const auto& ctx) { return ctx(); }) | ||
| | std::ranges::to<absl::flat_hash_set<ss::sstring>>(); | ||
| }; |
There was a problem hiding this comment.
I think for identity mapping we need to be slightly stricter than this. We should also validate that there is no context .ctx which doesn't exist in the source + exists in the destination + in scope in the source.
Unless I'm missing something, I think we won't need to consider the contents of the source SR at all, and these emptiness checks will just depend on the state of the config + destination schema registry.
There was a problem hiding this comment.
Done. Identity emptiness now reads the destination (list_subject_versions filtered by source scope) and never consults the source. An in-scope context present in the destination but absent from the source is now caught.
| } catch (...) { | ||
| eptr = std::current_exception(); | ||
| } | ||
| co_await client->shutdown(); |
There was a problem hiding this comment.
Can this shutdown throw? Wondering if we need to move it into the try-catch above
There was a problem hiding this comment.
Good catch — it can. Now wrapped in ss::coroutine::as_future, a failed close is logged and ignored so it can't mask the probe outcome.
…reation 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.
Two-cluster coverage of the SR preflight over validate_only and real create, positive and negative, asserting the specific failure message.
931d73a to
ce6f252
Compare
|
force-pushed: review from @pgellert and local Claude incorporated
|
CI test resultstest results on build#86761
|
Extends the shadow-link preflight (
manager::link_preflight_checks, whichalready probes the remote Kafka brokers) with two Schema Registry API-sync
checks. They run on the same path, so they gate both the
validate_onlydryrun and a real create:
settings, and credentials and list against it (
link_sr_unreachable).must be empty, scoped to the mapped contexts (
link_sr_target_not_empty).Both surface to the admin ConnectRPC as
FAILED_PRECONDITION.Backports Required
Release Notes