CORE-16766 Schema Migration: integrate client and test in ducktape end to end#30999
Open
pgellert wants to merge 18 commits into
Open
CORE-16766 Schema Migration: integrate client and test in ducktape end to end#30999pgellert wants to merge 18 commits into
pgellert wants to merge 18 commits into
Conversation
pgellert
requested review from
dotnwat and
nguyen-andrew
and removed request for
a team
pgellert
force-pushed
the
slsr/e2e-schema-sync
branch
from
79c0e2d to
ea78789
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
This PR integrates Schema Registry “API mode” sync into shadow-link cluster linking by introducing an HTTP-backed Schema Registry source reader, wiring it into the schema mirroring task, bootstrapping the destination _schemas topic on demand, and adding end-to-end ducktape coverage plus focused unit tests.
Changes:
- Add an HTTP REST-based Schema Registry
source_reader(plus unit tests) and wire it into the schema mirroring task, including reader reset on config changes and proper teardown. - Ensure the destination Schema Registry is bootstrapped (creates
_schemastopic) when any API-mode link exists, so syncing can start without external SR traffic. - Improve status/reporting semantics (proto
optionalpresence, start-time stamping) and add ducktape E2E tests covering sync, soft-deletes, error paths, backpressure, and leadership changes.
Reviewed changes
Copilot reviewed 28 out of 28 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| tools/type-checking/type-check-strictness.json | Adds the new ducktape test to the repo’s non-strict (“standard”) type-check list. |
| tests/rptest/tests/cluster_linking_schema_registry_sync_test.py | New ducktape E2E tests for SR API-mode sync scenarios (concurrency, soft deletes, recovery, backpressure, leadership transfer). |
| src/v/schema/registry.h | Extends schema::registry with ensure_started() to bootstrap internal SR state/topic. |
| src/v/schema/registry.cc | Implements ensure_started() for enabled/disabled registries. |
| src/v/schema/tests/fake_registry.h | Updates test fake to implement ensure_started(). |
| src/v/datalake/tests/record_schema_resolver_test.cc | Updates test registry implementation to implement ensure_started(). |
| src/v/redpanda/admin/services/shadow_link/tests/converter_test.cc | Updates/extends admin converter tests for presence-aware SR sync status fields and adds an “idle/never-synced” regression test. |
| proto/redpanda/core/admin/v2/shadow_link.proto | Makes SR sync status and timestamps presence-aware via optional fields. |
| src/v/pandaproxy/schema_registry/rest_client/BUILD | Exposes rest_client libs to cluster_link subpackages for the new HTTP reader. |
| src/v/cluster_link/utils.h / src/v/cluster_link/utils.cc | Factor TLS file-or-inline conversion helpers for reuse (CA, key/cert). |
| src/v/cluster_link/service.h / src/v/cluster_link/service.cc | Adds destination SR bootstrapping when SR API-mode links exist and wires in HTTP source reader factory. |
| src/v/cluster_link/schema_registry_sync/source_reader.h | Extends source-reader API with per-link config in factory, adds subject_not_found, and adds a stop() hook. |
| src/v/cluster_link/schema_registry_sync/unavailable_source_reader.{h,cc} | Improves “unavailable” reader to carry a cause-specific message and updates factory signature. |
| src/v/cluster_link/schema_registry_sync/mirroring_task.{h,cc} | Resets reader on config change, stops reader on task stop, syncs destination inventory on refresh, and handles fully-soft-deleted subjects correctly. |
| src/v/cluster_link/schema_registry_sync/http_source_reader.{h,cc} | New HTTP-backed Schema Registry reader implementation and config parsing. |
| src/v/cluster_link/schema_registry_sync/tests/http_source_reader_test.cc | New unit tests validating HTTP reader behavior, error mapping, and URL parsing. |
| src/v/cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h | Aligns fake source-reader behavior with real SR (404 on active-only fully soft-deleted subject), updates factory signature, and forwards ensure_started(). |
| src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc | Asserts cumulative start-time is now stamped. |
| src/v/cluster_link/schema_registry_sync/tests/BUILD | Adds the new http_source_reader_test target. |
| src/v/cluster_link/schema_registry_sync/BUILD | Adds the new http_source_reader library target and adjusts deps. |
| src/v/cluster_link/BUILD | Adds deps for net TLS and the HTTP reader integration. |
Collaborator
dotnwat
previously approved these changes
Jul 2, 2026
Comment on lines
+80
to
+81
| std::from_chars(port_sv.data(), port_sv.data() + port_sv.size(), port) | ||
| .ec |
Member
There was a problem hiding this comment.
i think there is also technically another condition regarding .ptr that is used to define success in addition to the .ec value.
| // Sync currently running. This is unset when no Schema Registry sync is | ||
| // running. | ||
| SchemaRegistryCurrentSync current_sync = 2; | ||
| optional SchemaRegistryCurrentSync current_sync = 2; |
Member
There was a problem hiding this comment.
i thought in newer protobuf everything was optional by default?
Contributor
Author
There was a problem hiding this comment.
This is a quirk of our custom protobuf codegen. It only generates presence tracking for optional fields. They are wire-compatible, though.
| ss::abort_source& as) { | ||
| // Destination/internal faults bubble out and become `faulted`. The scan | ||
| // sinks the predicate by value; forward the borrowed one. | ||
| // list_subject_versions doesn't sync; force it so the inventory isn't |
Member
There was a problem hiding this comment.
what does list_subject_versions have to do with this function? is it a caller?
Contributor
Author
There was a problem hiding this comment.
I've reworked the code now to make it more obvious, it's from within scan_destination_inventory
nguyen-andrew
left a comment
nguyen-andrew
left a comment
Member
There was a problem hiding this comment.
🆒 nice job, esp with the tests
| template<typename T1, typename T2> | ||
| requires(!std::is_same_v<T1, T2>) | ||
| net::key_store operator()(const T1&, const T2&) { | ||
| vunreachable("TLS key and cert must be of the same type"); |
Member
There was a problem hiding this comment.
Not new in this PR, but wondering if vunreachable too strong here. Not sure if it's "impossible" for them to not be of the same type or not - but fine with leaving it.
Contributor
Author
There was a problem hiding this comment.
It's unreachable because of the structure of the TLSSettings we use:
redpanda/proto/redpanda/core/common/v1/tls.proto
Lines 25 to 65 in 96ee14f
|
|
||
| // GET /subjects returns subjects across all contexts; the reader keeps only the | ||
| // requested context so discovery stays single-context. | ||
| TEST(http_source_reader, list_subjects_filters_to_requested_context) { |
Member
There was a problem hiding this comment.
I'm assuming that in the future we can call list subjects with a specific context (instead of getting all subjects across all contexts and doing client-side filtering)?
Contributor
Author
There was a problem hiding this comment.
Good point, I've implemented support for this now in the rest_client
| // Stamp the cumulative summary's start time once per task instance (the | ||
| // proto documents totals_since_task_start.start_time as the task start). | ||
| // A fresh instance after a leadership change re-stamps it on its first run. | ||
| if (!_status.totals_since_task_start.start_time.has_value()) { |
Member
There was a problem hiding this comment.
Does anything reset _status/_last_full_sync when the same task instance regains leadership later (A -> B -> A)?
Contributor
Author
There was a problem hiding this comment.
Good spot, now we do. Fixed + added a test
| } // namespace | ||
|
|
||
| std::optional<net::unresolved_address> | ||
| parse_source_address(std::string_view url) { |
Member
There was a problem hiding this comment.
Would this work with a url with a path prefix (e.g. https://proxy.example.com/schema-registry)? Would path prefixes be something we should support?
Contributor
Author
There was a problem hiding this comment.
No, it wouldn't, and I don't believe we need to support them. Our implementation assumes that we sync the full schema registry, so it's not like you can point it at a /contexts/{context} path and sync just that.
| source_error to_source_error(rc::domain_error err) { | ||
| auto kind = source_error_kind::operation_failed; | ||
| if ( | ||
| const auto* call = std::get_if<rc::http_call_error>(&err); |
Member
There was a problem hiding this comment.
Are there any specific HTTP error codes that we should handle differently? E.g., should auth failures be considered operation_failed or source_unavailable? I don't think it matters much, just a question.
Contributor
Author
There was a problem hiding this comment.
Yeah, I think eagerly calling authN/authZ failures as unavailable makes sense. These will mostly surface when someone is first configuring a link, and treating these as link unavailable and failing the whole link seems like better UX here. I've changed to that now.
| // internally, and a persistent failure must not fail link handling. | ||
| auto started = co_await ss::coroutine::as_future( | ||
| _schema_registry_dest->ensure_started()); | ||
| if (started.failed()) { |
Member
There was a problem hiding this comment.
Since this is marked as best-effort, what are the consequences of this not succeeding - how would the task handle a _schemas topic missing? If this happened, would it self heal, or would an operator have to take an action to heal it?
Contributor
Author
There was a problem hiding this comment.
An operator action would be needed to unblock (e.g. send a request to the destination schema registry / restart the broker / recreate the link) because the mirror task only runs on the _schemas topic leader shard, so it doesn't run unless there is a schemas topic.
mnajda-redpanda
previously approved these changes
Jul 3, 2026
Comment on lines
+51
to
+61
| if ( | ||
| const auto* call = std::get_if<rc::http_call_error>(&err); | ||
| call != nullptr && std::holds_alternative<ss::sstring>(*call)) { | ||
| kind = source_error_kind::source_unavailable; | ||
| } else if (std::holds_alternative<rc::retries_exhausted>(err)) { | ||
| kind = source_error_kind::source_unavailable; | ||
| } else if (std::holds_alternative<rc::aborted_error>(err)) { | ||
| kind = source_error_kind::source_unavailable; | ||
| } else if (std::holds_alternative<rc::subject_not_found>(err)) { | ||
| kind = source_error_kind::subject_not_found; | ||
| } |
Contributor
There was a problem hiding this comment.
I guess ss::visit would be cleaner here?
Contributor
Author
|
/ci-repeat 5 |
Collaborator
Retry command for Build#86704please wait until all jobs are finished before running the slash command |
pgellert
force-pushed
the
slsr/e2e-schema-sync
branch
from
ea78789 to
4541296
Compare
The source-reader factory's create() took no arguments and the only implementation returned the unavailable reader. The upcoming HTTP-backed reader needs the link's Schema-Registry-API connection settings (source URL, auth, TLS) to build its transport, so thread the api-mode config pointer through create() from the task's already-resolved _config. No behavior change: the registered factory is still the unavailable reader, which ignores the argument, and api_mode() is null outside SR-API mode.
create_tls_configuration's key/cert-pair visitor and CA-certificate conversion are generic over model::tls_file_or_value, not specific to the Kafka-side connection_config. Export them from utils.cc as to_certificate/to_key_store so the upcoming HTTP Schema Registry source reader can reuse them instead of re-deriving the same conversion.
Add an optional context parameter to list_subjects that sends a subjectPrefix=":<ctx>:" query hint and filters the response to that context client-side, mirroring list_contexts's contextPrefix handling. This lets a caller discover the subjects in one context without pulling every context and filtering itself. The client-side filter keeps the result correct against a server that ignores the hint (Redpanda's own server honors it).
Adapt the Schema Registry REST client to the source_reader interface so the shadowing task can read a remote source registry over HTTP(S). The reader maps the rest_client's typed errors onto source_error (reachable source -> operation_failed; an unreachable one -> source_unavailable) and serializes requests onto one in-flight slot (a single http::client cannot service concurrent requests). Subjects are always read in qualified form; unqualified sources are out of scope for now. The factory resolves the connection synchronously from the link's API-mode config (URL, basic auth, TLS material) but builds the http::client lazily on first use, because constructing TLS credentials is async while the factory is not; a null config or unparseable URL yields the unavailable reader so the link parks. The source URL is parsed with ada rather than by hand, and the TLS certificate/key-store conversion reuses cluster_link::to_certificate/to_key_store rather than re-deriving it for this second client. Contexts are enumerated over the rest_client's GET /contexts, so subjects in non-default contexts are discovered. Also adds a source_reader::stop() hook (so a reader can release its transport) and widens the rest_client targets to cluster_link. Not yet registered -- the next commit tests it.
Drive the reader over a mocked HTTP transport: list_contexts enumerating every source context, single-context filtering of GET /subjects, schema fetch, and the source_error classification that decides whether the link parks (unreachable -> source_unavailable) or skips an item (a reachable error -> operation_failed). Also checks the factory parks on a null or unparseable config.
Register the HTTP-backed source reader in place of the unavailable one, so an API-mode shadow link now reads its source registry over HTTP. Add a transport lifecycle on the task: stop() releases the reader once the run loop has stopped (no fiber is using it then), and a config change rebuilds the reader before the next run so a new source URL, auth, or TLS setting takes effect. The rebuild runs at the top of run_impl, after the config-changed flag is consumed and before any reconcile fiber starts.
The registry ensures the Schema Registry has finished starting -- which creates its internal _schemas topic and loads the schema store -- before each store access, but only privately, on the leader-gated import/sync path. Expose a public method that performs just the topic-creation step, so a caller can bootstrap a destination Schema Registry without also forcing the store load: that load is redundant off the _schemas/0 leader, and its leader-dependent retry loop must not run eagerly on every node. Shadow-link SR-API sync (next commit) needs this: its task runs only on the _schemas/0 partition leader, but the Schema Registry creates that topic lazily on the first client request, so the topic must be created out-of-band for the task to ever start. The store load itself is then triggered by the task's first destination operation, on the leader. Idempotent, and a no-op when the Schema Registry is disabled. Every implementer, including the test doubles, gains the method.
An SR-API-mode link's sync task runs only on the destination's _schemas/0 leader, but the Schema Registry creates _schemas lazily on first client access -- so on a never-used destination the topic never exists, no leader is elected, and the sync silently does nothing. Bootstrap the destination with registry::ensure_internal_topic() for api-mode links present at manager start and created later (via the link-change notification). Gate on an api-mode link existing, not on shadow linking being enabled, so a cluster without one never creates _schemas. Topic creation is cluster-wide, so it runs on a single shard only. The full start-up (store load) is deliberately not run here: it happens on the _schemas/0 leader when the sync task first touches the destination, so eager per-node loads cannot park shutdown behind the Schema Registry's leader-dependent start-up retries.
The admin-v2 codegen (bazel/pbgen) only tracks field presence for oneof
or proto3 optional fields; a plain singular field is serialized
unconditionally. So SchemaRegistrySyncStatus.current_sync/last_full_sync
and SchemaRegistrySyncSummary.start_time/finish_time were emitted even
when unset -- an unset message as {}, an unset Timestamp as the epoch --
contradicting their "unset when ..." doc comments. current_sync being
always-present also made its sync_type read UNSPECIFIED when idle.
Mark the four presence-bearing fields optional so the codegen tracks
has_x() and the converter's existing has_value() gating reaches the wire.
No converter change; inventory/totals stay always-present. Adding optional
to a proto3 struct field is wire-compatible.
This does change proto3 JSON output (an unset field is now omitted rather
than serialized as its default), but rpk always talks to admin-v2 over
Connect's binary protobuf codec -- it never requests connect's JSON wire
encoding for this or any other admin-v2 service -- so rpk needs no change.
Replicate a reference-heavy schema set from a source cluster's Schema Registry to a destination cluster's over the shadow-link HTTP source reader. Two clusters via ShadowLinkTestBase, an SR-API-mode link, and a layered diamond DAG (leaves -> mids -> tops, shared referents) at the scale of the reconciler concurrent_stress unit test, so the concurrent fetch/import path runs against a real registry. Covers additions racing the first in-flight sync and additions after a completed sync (a new version, a standalone subject, new tops), checking every (subject, version) against the source's body, references, and preserved schema ID, then the SchemaRegistrySyncStatus counters via both the admin API and rpk shadow status. Plaintext, default context, no auth.
Add an end-to-end ducktape test asserting that a source version's soft-delete state converges on the destination over the SR HTTP API: a version soft-deleted before the first full sync, and a version soft-deleted after a completed sync. The task diffs the active vs. deleted-inclusive listings to select soft-deleted versions for import (the bare version listing carries no per-version deleted flag), and read_subject_version fetches each body with deleted=true, so the single-version GET returns the deleted flag, which the writer honours on import. This is the end-to-end guard the create-only tests lacked.
A subject soft-deleted in whole reports subject-not-found on its active-only version listing, so discovery skipped it (and counted a spurious per-item error), leaving the destination's versions active rather than converging to deleted. The deleted-inclusive listing still returns the versions, so they are recoverable. Add a subject_not_found source_error_kind (mapped from the rest_client), list the deleted-inclusive versions first, and treat a not-found listing as an empty result rather than a fault -- so a fully soft-deleted subject's versions are discovered and re-imported as deleted. A transient or other failure is still propagated or counted, never read as "no active versions" (which would wrongly delete live destination versions). Covered by an e2e whole-subject-delete-after-sync case, an http_source_reader unit test for the 404->subject_not_found mapping, and the reconciler fixture's fully-soft-deleted subject.
Every existing e2e schema is valid, so the error counters were never driven non-zero end-to-end. Scope the link to a referrer plus an independent subject but exclude the referrer's referent: importing the referrer fails on the unresolvable reference and is counted, while the independent subject still syncs. Assert last_full_sync.errors >= 1 and totals_since_task_start.errors >= 1 -- the error is recorded within a completed full sync, so the link keeps syncing rather than parking or faulting.
Point a new link at an unreachable endpoint so the task parks without completing a sync (last_error_message set, last_full_sync unset), then repoint it at the real source via update_link. The config change forces a fresh full sync that reaches the source and replicates, exercising both the unavailable-then-recover path and config-driven resync.
Shrink the destination's schema_registry_sync_memory_bytes to its 1 MiB floor and raise schema_registry_sync_parallelism to 14, then seed 16 ~100 KiB schema bodies so the concurrently-fetched bodies (~1.4 MiB) exceed the budget, forcing the byte-semaphore to serialize. The sync must still complete -- the real-registry counterpart to the unit deadlock canary, confirming progress under backpressure with actual bytes on the wire.
The sync task runs only on the _schemas/0 leader and holds its status in memory, so a leadership move is the real test of whether counters are re-derived from durable state rather than phantom-accumulated. Sync a DAG on the first instance, transfer _schemas/0 leadership to another broker, and assert the new instance: starts fresh (later task start_time), rebuilds the destination inventory to the full count, completes a full sync that imports nothing (last_full_sync.subject_versions_changed == 0), and then still imports a subject added after the bounce. This surfaced a real bug: list_subject_versions reads the local store as-is, so a freshly-elected leader could see a stale view of its own _schemas replay, spuriously re-import already-present versions, and permanently inflate totals_since_task_start (it only ever accumulates, so the counter would never recover within that task instance). Fix refresh_destination_inventory to sync the destination before scanning.
totals_since_task_start.start_time is documented as the time the Schema Registry sync task started, but the task only ever stamped current_sync.summary.start_time -- the cumulative summary's start_time was left unset, so the admin API and rpk reported it as empty. Stamp it once per task instance on the first run. A fresh instance after a leadership change re-stamps it, so the field reflects the current leader's task. A mirroring_task unit test asserts the field is set after a full sync, and the e2e asserts it is present after a sync and carries a later value on the post-bounce instance.
The mirroring task instance is created once per link and reused across leadership changes: losing leadership only stops the runner, so its in-memory status, destination inventory, and last-full-sync timestamp survived. On an A->B->A handoff the returning leader would report a prior tenure's stale counters and could skip its first full sync while _last_full_sync was still recent, deferring catch-up by up to the full-sync interval. Reset the derived state in stop() (safe there: task::stop() has closed the runner gate, so no run_impl is in flight). _config and the pending _config_changed flag are preserved, so a config change queued while stopped still takes effect. Every leadership acquisition now re-derives from the durable destination store, matching a genuinely fresh instance. Extend the leadership-change e2e to bounce _schemas/0 leadership back to the original broker and assert the reused instance also starts fresh: a later start_time, zeroed totals, and a full sync that imports nothing.
pgellert
force-pushed
the
slsr/e2e-schema-sync
branch
from
4541296 to
ff5256d
Compare
Contributor
Author
|
force-push:
|
7 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
See the commit messages for details.
Fixes https://redpandadata.atlassian.net/browse/CORE-16766
Backports Required
Release Notes